【Redis】之 RedLock 分布式锁

本文介绍: 在多线程环境中，为了控制线程对资源并发访问和竞争，我们经常需要用到锁来进行控制。常用的锁例如 Jav a 自带的等。但这些锁只能用于单机系统中，如果涉及到多机器、多节点的分布式环境的资源竞争，就需要使用分布式锁了。

一、Re dLoc k 详解

1、什么是 Re dLock

在了解 Re dLock 之前，我们需要先了解一下分布式锁的原理：【Redis】之分布式锁

简单来说就是 Re dLock 是 Redi s 实现分布式锁的一种方式。但不同点在于 RedLock 是 Redi s 的作者 Anti rez 在单 Redi s 节点基础上引入的高可用模式。

2、为什么要使用 RedLock

在【Redis】之分布式锁中我也指出了，不管使用原生的 SET key value EX NX 命令还是使用 Redi s son 这个能有效解决锁续期的方案，它们都无法解决 Redi s 在主从复制、哨兵集群下的多节点问题：

客户端 A 将 Ke y 写入到 Mast e r 节点成功获取到锁；
此时 Mast e r 节点发生故障，Ke y 没有来得及同步到 Slave 上（数据是后台通过异步同步的）；
Slave 节点升级为 Master 节点；
客户端 B 从新的 Master 节点获取到了对应同一个资源的锁。

这种情况下锁的安全性被打破了，所以 Redi s 的作者就实现了解决这种问题的 RedLock 锁。

3、RedLock 加锁 原理

在 Redi s 分布式集群中，假设我们有 5 个 Redi s 节点（中小规模项目一般是：1主4从+3哨兵），则 RedLock 加锁过程如下：

获取当前Uni x 时间，以毫秒为单位，并设置锁的超时时间 TTL（TTL 时间要大于正常业务执行的时间 + 成功获取锁消耗的时间 + 时钟漂移）；
依次从 5 个节点中获取锁，需要使用相同的 key 和具有唯一性的 value。获取锁时，需要设置一个网络连接和响应超时时间，这个超时时间要小于锁的失效时间 TTL，从而避免客户端死等。比如：TTL 为 5s，设置获取锁最多用1s，所以如果一秒内无法获取锁，就放弃获取这个锁，尝试从下个节点获取锁；
客户端获取所有能获取的锁后的时间减去第 1 步的时间，就得到了获取锁消耗的时间（锁的获取时间要小于锁的失效时间 TTL，并且至少从半数以上 (N/2+1)的Redis节点取到锁，才算获取成功锁）；
成功获得锁后，key 的真正有效时间 = TTL – 锁的获取时间 – 时钟漂移。比如：TTL 是10s，获取所有锁用了 2s，则真正锁有效时间为 8s；
如果获取锁失败（没有在半数以上实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的 Redis 实例上进行解锁，即使是没有加锁成功的 Redis 实例；
失败重试：当客户端获取锁失败时，应该在随机时间后重试获取锁；同时重试获取锁要有一定次数限制（在随机时间后进行重试，主要是防止过多的客户端同时尝试去获取锁，导致彼此都获取锁失败的问题）；

加锁失败的实例也要执行解锁操作的原因是：可能会出现服务端响应消息丢失但实际上成功了的情况。

设想这样一种情况：客户端发给某个 Redis 节点的获取锁的请求成功到达了该 Redis 节点，这个节点也成功执行了 SET 操作，但是它返回给客户端的响应包却丢失了。这在客户端看来，获取锁的请求由于超时而失败了，但在Redis这边看来，加锁已经成功了。因此，释放锁的时候，客户端也应该对当时获取锁失败的那些Redis节点同样发起请求。实际上，这种情况在异步通信模型中是有可能发生的：客户端向服务器通信是正常的，但反方向却是有问题的。

4、RedLock 崩溃 恢复问题

由于 N 个 Redis 节点中的大多数能正常工作就能保证 Redlock 能正常工作，因此理论上它的可用性更高。所以前面我们说的主从架构下存在的安全性问题，在 RedLo ck 中已经不存在了。但如果所有节点同时发生崩溃重启的情况下还是会对锁的安全性有影响，具体的影响程度跟 Redis 持久化配置有关：

假如我们没有开启 Redis 的持久化功能（比如 AOF），在 ClientA 获取锁成功后，所有 Redis 实例重启后，ClientB 也能够再次获取到锁，这样违法了锁的排他互斥性；
但实际上我们一般都会启动 AOF 持久化功能，即使如此也会存在一个问题：但是由于 AOF 同步到磁盘的方式默认是每秒一次，所以如果在一秒内断电，则会导致数据丢失，立即重启后同样会造成锁互斥性的失效；
但如果同步磁盘方式使用 Always（每一个写命令都同步到硬盘），虽然避免了数据丢失问题，但会造成性能急剧下降，所以需要在锁完全有效性和性能方面进行取舍。

解决方案：

为了有效解决既保证锁完全有效性有保证 Redis 性能高效的问题，Redis 的作者 antirez 提出了 延迟重启 的概念：Redis 同步到磁盘方式保持默认的每秒1次，在 Redis 崩溃后（无论是一个还是所有），先不立即重启它，而是等待 TTL 时间后再重启。

这样的话，这个节点在重启前所参与的锁都会过期，它在重启后就不会对现有的锁造成影响，但缺点是在TTL时间内服务相当于暂停状态。但这只是一种参考方案，需要根据实际来判断是否使用。

5、RedLock 的弊端

虽然 RedLock 解决 Redis 集群部署下的安全性问题，但同时也牺牲了性能：原来只要主节点写成功了就行了，现在要很多台机器成功加锁才算加锁成功；
如上面所示，当 Redis 全部重启，由于持久化的问题，RedLock 也会存在锁失效的问题，所以 RedLock 并不能100%解决锁失效问题。

二、RedLock 实战

1、基于 Redisson 的 RedLock 实现

在 JAVA 的 Redisson 包中基于 Redis 的 Redisson 红锁 RedissonRedLock 对象实现了Redlock 介绍的加锁算法。该对象也可以用来将多个 RLock 对象关联为一个红锁，每个RLock 对象实例可以来自于不同的 Redisson实例。

RLock lock1 = redissonInstance1.getLock("lock1");
RLock lock2 = redissonInstance2.getLock("lock2");
RLock lock3 = redissonInstance3.getLock("lock3");

RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
// 同时加锁：lock1 lock2 lock3
// 红锁在大部分节点上加锁成功就算成功。
lock.lock();
...
lock.unlock();

大家都知道，如果负责储存某些分布式锁的某些Redis节点宕机以后，而且这些锁正好处于锁住的状态时，这些锁会出现锁死的状态。为了避免这种情况的发生，Rediss on 内部提供了一个监控锁的看门狗，它的作用是在 Rediss on 实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改Config.lockWatchdogTimeout 来另行指定。

关于 Rediss on 看门狗的机制原理可以参考我的另一篇博客：【Redis】之 Redisson 分布式锁

另外 Redisson 还通过加锁的方法提供了 leaseTime 的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
// 给lock1，lock2，lock3加锁，如果没有手动解开的话，10秒钟后将会自动解开
lock.lock(10, TimeUnit.SECONDS);

// 为加锁等待100秒时间，并在加锁成功10秒钟后自动解开
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
...
lock.unlock();

除了 RedLock 之外，Redisson 还封装了可重入锁（Reentrant Lock）、公平锁（Fair Lock）、读写锁（ReadWriteLock）、信号量（Semap hore）等，具体使用说明可以参考官方文档：Redisson的分布式锁和同步器

2、RedLock 实现 原理

Redisson 中的 RedLock 继承于 Mul tiLock（RedissonRedLock extends RedissonMul tiLock），所以 redLock.tryLock 实际调用org.redisson.RedissonMultiLock.java#tryLock()，进而调用到其同类的 tr yLock(long waitTime, long leaseTime, TimeUnit unit) 源码如下：

final List<RLock&gt; locks = new ArrayList<&gt;();

/**
 * Creates instance with multiple {@link RLock} objects.
 * Each RLock object could be created by own Redisson instance.
 *
 * @param locks - array of locks
 */
public RedissonMultiLock(RLock... locks) {
	if (locks.length == 0) {
		throw new IllegalArgumentException("Lock objects are not defined");
	}
	this.locks.addAll(Arrays.asList(locks));
}

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    long newLeaseTime = -1;
    if (leaseTime != -1) {
        newLeaseTime = unit.toMillis(waitTime)*2;
    }
    
    long time = System.currentTimeMillis();
    long remainTime = -1;
    if (waitTime != -1) {
        remainTime = unit.toMillis(waitTime);
    }
    long lockWaitTime = calcLockWaitTime(remainTime);
    /**
     * 1. 允许加锁失败节点个数限制（N-(N/2+1)）
     */
    int failedLocksLimit = failedLocksLimit();
    /**
     * 2. 遍历所有节点通过EVAL命令执行lua加锁
     */
    List<RLock> acquiredLocks = new ArrayList<>(locks.size());
    for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
        RLock lock = iterator.next();
        boolean lockAcquired;
        /**
         *  3.对节点尝试加锁
         */
        try {
            if (waitTime == -1 &amp;&amp; leaseTime == -1) {
                lockAcquired = lock.tryLock();
            } else {
                long awaitTime = Math.min(lockWaitTime, remainTime);
                lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
            }
        } catch (RedisResponseTimeoutException e) {
            // 如果抛出这类异常，为了防止加锁成功，但是响应失败，需要解锁所有节点
            unlockInner(Arrays.asList(lock));
            lockAcquired = false;
        } catch (Exception e) {
            // 抛出异常表示获取锁失败
            lockAcquired = false;
        }
        
        if (lockAcquired) {
            /**
             *4. 如果获取到锁则添加到已获取锁集合中
             */
            acquiredLocks.add(lock);
        } else {
            /**
             * 5. 计算已经申请锁失败的节点是否已经到达 允许加锁失败节点个数限制 （N-(N/2+1)）
             * 如果已经到达， 就认定最终申请锁失败，则没有必要继续从后面的节点申请了
             * 因为 Redlock 算法要求至少N/2+1 个节点都加锁成功，才算最终的锁申请成功
             */
            if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
                break;
            }

            if (failedLocksLimit == 0) {
                unlockInner(acquiredLocks);
                if (waitTime == -1 &amp;&amp; leaseTime == -1) {
                    return false;
                }
                failedLocksLimit = failedLocksLimit();
                acquiredLocks.clear();
                // reset iterator
                while (iterator.hasPrevious()) {
                    iterator.previous();
                }
            } else {
                failedLocksLimit--;
            }
        }

        /**
         * 6.计算 目前从各个节点获取锁已经消耗的总时间，如果已经等于最大等待时间，则认定最终申请锁失败，返回false
         */
        if (remainTime != -1) {
            remainTime -= System.currentTimeMillis() - time;
            time = System.currentTimeMillis();
            if (remainTime <= 0) {
                unlockInner(acquiredLocks);
                return false;
            }
        }
    }

    if (leaseTime != -1) {
        List<RFuture<Boolean>> futures = new ArrayList<>(acquiredLocks.size());
        for (RLock rLock : acquiredLocks) {
            RFuture<Boolean> future = ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);
            futures.add(future);
        }
        
        for (RFuture<Boolean> rFuture : futures) {
            rFuture.syncUninterruptibly();
        }
    }

    /**
     * 7.如果逻辑正常执行完则认为最终申请锁成功，返回true
     */
    return true;
}