Redisson 实现RedLock详解

要实现分布式锁，Redis官网介绍了三个必需要保证的特性：html

安全特性：互斥。任意时刻都只能有一个客户端可以持有锁。
活跃性A：无死锁。即便在持有锁的客户端崩溃，或者出现网络分区的状况下，依然可以获取锁。
活跃性B：容错。只要多数Redis节点是存活状态，客户端就能申请锁或释放锁。

为何基于故障转移的分布式锁还不够呢

咱们分析一个场景：redis

1.客户端A在Redis master节点申请锁算法

2.master在将存储的key同步到slave上以前崩溃了安全

3.slave晋升为mastermarkdown

4.客户端B申请一个客户端A已经持有的资源的锁网络

哇，问题出现了。分布式

Redlock算法

为了解决基于故障转移实现的分布式锁的问题，Redis做者提出了Redlock算法。ide

假设有N（N≥5）个Redis的master实例，这些节点之间都是相互独立的，所以咱们不用副本或其余隐式的协调系统。为了获取锁，客户端须要进行如下操做（假设N=5）：函数

1.获取当前时间oop

2.按顺序尝试在5个Redis节点上获取锁，使用相同的key 做为键，随机数做为值（随机值在5个节点上是同样的）。在尝试在每一个Redis节点上获取锁时，设置一个超时时间，这个超时时间须要比总的锁的自动超时时间小。例如，自动释放时间为10秒，那么链接超时的时间能够设置为5-50毫秒。这样能够防止客户端长时间与处于故障状态的Redis节点通讯时保持阻塞状态：若是一个Redis节点处于故障状态，咱们须要尽快与下一个节点进行通讯。

3.客户端计算获取锁时消耗的时间，用当前时间，减去在第1步中获得的时间。只用当客户端能够在多数节点上可以获取到锁，而且获取锁消耗的总时间小于锁的有效时间，那么这个锁被认为是获取成功了。

4.若是锁获取成功了，锁的有效时间是初始的有效时间减掉申请锁消耗的总时间。

5.若是客户端申请锁失败了(例如不知足多数节点或第4步中获取到的锁有效期为负数)，客户端须要在全部节点上解除锁（即便是认为已经没法提供服务的节点）。

当一个客户端没法获取锁时，它应该在随机延迟后再次尝试，以便让多个客户端在同一时间尝试获取相同资源的锁（可能会致使脑裂的状况），并且，在大多数Redis实例中，客户端尝试获取锁定的速度越快，出现裂脑状况的窗口就越小，所以，理想状况下，客户端应尝试使用多路复用将SET命令发送到N个实例。

值得强调的是，对于未能获取大多数锁的客户端，尽快释放（部分）获取的锁，这是多么重要，所以，无需等待key到期便可再次获取锁（可是，若是发生网络分区，而且客户端再也不可以与Redis实例通讯，则在等待key到期时须要支付可用性损失）。

有细心的同窗可能会有疑惑，在每一个节点上设置的锁时间是同样的，可是在每一个节点执行的时间可能不同，那锁失效的时间也有早有晚啊，那这个算法够安全吗？咱们一块儿来分析一下不一样场景下的状况：

首先假设客户端在每一个节点上都能获取到锁，每一个节点上都有相同的存活时间的key。可是，key在每一个节点上设置的时间点是不同的，那么key的过时时间就不同，假设第一个节点上设置锁的时间点为T1,最后一个节点设置锁的时间点为T2，那么第一个锁的有效期至少为

MIN_VALIDITY=TTL-(T2-T1)-CLOCK_DRIFT(每一个计算机都有一个本地时间，而每一个计算机可能都存在很是小的时间偏差，这个时间偏差即为CLOCK_DRIFT)，其余节点上的key都会在这个时间以后才会到期，因此咱们能够肯定至少在这一时间内，全部的key是都生效的。

在一个客户端已经在多数节点上设置了key的这段时间内，其余的客户端是没法再在多数节点上（N/2+1）获取到锁的。

咱们还须要保证多个客户端在同一时间不能获取到同一个锁。

若是一个客户端在多数节点上获取锁所消耗的时间接近或者大于锁的有效期(TTL),咱们认为这个锁没有申请成功，并会在全部节点上进行解锁操做，因此咱们只要考虑加锁消耗时间小于锁有效期的状况便可，在这种状况下，对于上面已经说明的参数——MIN_VALIDITY，没有客户端可以从新获取锁。因此只有在在多数节点上（N/2+1）加锁耗时大于TTL的状况下，才会有多个客户端可能会同时获取到锁，这种状况直接将锁认定为获取失败便可。

这套算法的可用性是基于如下三个特性：

1.自动释放锁——锁能够自动失效，失效后能够从新上锁

2.一般状况下，客户端一般会在未得到锁或得到锁且工做终止时删除锁，这使得咱们没必要等待key过时便可从新得到锁。

3.当客户端须要重试获取锁时，它等待的时间要比获取大多数节点锁定所需的时间长得多，以便几率地使资源争用期间的脑裂状况变得不可能。

然而，咱们付出了时间为TTL的网络分区可用性的代价，若是发生了连续性的分区，那咱们也要付出无限的分区可用性代价（CAP理论，一致性、可用性、分区容错性，分区一致性是前提）。每当客户端得到一个锁并在可以删除锁以前被分区时，都会发生这种状况。（以下图所示，有三个节点的分区不可用的状况下，client A 和client B同时申请一个锁，这时两个客户端都没法获取到多数节点，那么一直到网络恢复或者锁的超时时间，竞争关系解除，从新再竞争...）

Redisson的实现

以上就是Redis的做者给出的Redlock的算法模型，那么在Java的Redis客户端中，Redisson实现了Redlock,咱们来分析一下它的具体实现代码。

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不只提供了一系列的分布式的Java经常使用对象，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最简单和最便捷的方法。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者可以将精力更集中地放在处理业务逻辑上。

按照惯例，咱们先来看一下源码

/**
 * RedLock locking algorithm implementation for multiple locks. 
 * It manages all locks as one.
 * 
 * @see <a href="http://redis.io/topics/distlock">http://redis.io/topics/distlock</a>
 *
 * @author Nikita Koksharov
 *
 */
public class RedissonRedLock extends RedissonMultiLock {

    /**
     * Creates instance with multiple {@link RLock} objects.
     * Each RLock object could be created by own Redisson instance.
     *
     * @param locks - array of locks
     */
    public RedissonRedLock(RLock... locks) {
        super(locks);
    }

    @Override
    protected int failedLocksLimit() {
        return locks.size() - minLocksAmount(locks);
    }
    
    protected int minLocksAmount(final List<RLock> locks) {
        return locks.size()/2 + 1;
    }

    @Override
    protected long calcLockWaitTime(long remainTime) {
        return Math.max(remainTime / locks.size(), 1);
    }
    
    @Override
    public void unlock() {
        unlockInner(locks);
    }

}
复制代码

经过构造函数咱们能够看出，要构造出一个RedissonRedLock,须要至少一个RLock实例，具体实现须要到父类RedissonMultiLock中查看，父类实际上是将构造函数传入的RLock添加到了一个List的列表中。至于什么是RLock呢，这是Redisson中对锁的最高层的抽象，它的实现类包括RedissonWriteLock、RedissonReadLock、RedissonFairLock，固然咱们正在分析的RedissonMultiLock也是它的实现类。

假设咱们有三个Redis节点，

Config config1 = new Config();
config1.useSingleServer().setAddress("redis://localhost:5378")
        .setPassword("").setDatabase(0);
RedissonClient redissonClient1 = Redisson.create(config1);
 
Config config2 = new Config();
config2.useSingleServer().setAddress("redis://localhost:5379")
        .setPassword("").setDatabase(0);
RedissonClient redissonClient2 = Redisson.create(config2);
 
Config config3 = new Config();
config3.useSingleServer().setAddress("redis://localhost:5380")
        .setPassword("").setDatabase(0);
RedissonClient redissonClient3 = Redisson.create(config3);
 
String lockKey = "REDLOCK";
RLock lock1 = redissonClient1.getLock(lockKey);
RLock lock2 = redissonClient2.getLock(lockKey);
RLock lock3 = redissonClient3.getLock(lockKey);


RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
try {
    if (redLock.tryLock(10, 5, TimeUnit.SECONDS)) {
        //TODO if get lock success, do something;
    }
 }catch(Exception e){
    
 }
复制代码

RedissonRedLock彻底的按照上文咱们介绍的Redlock的算法来实现的，经过在三个不一样节点上分别获取锁，来构造一个Redlock,咱们再来分析一下具体的tryLock的实现，这个方法是在RedissonRedLock的父类RedissonMultiLock实现的：

/**
     *
     * @param waitTime the maximum time to acquire the lock
     * @param leaseTime lease time
     * @param unit time unit
     * @return
     * @throws InterruptedException
     */
    @Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long newLeaseTime = -1;
        if (leaseTime != -1) {
            if (waitTime == -1) {
                newLeaseTime = unit.toMillis(leaseTime);
            } else {
                //等待时间和加锁时间都不为-1时，newLeaseTime为waitTime时间的两倍
                newLeaseTime = unit.toMillis(waitTime)*2;
            }
        }
        
        long time = System.currentTimeMillis();
        long remainTime = -1;
        if (waitTime != -1) {
            remainTime = unit.toMillis(waitTime);
        }
        //子类重写的方法，Math.max(remainTime / locks.size(), 1)，waitTime除以节点的个数，与1取较大值
        //Math.max(等待时间的毫秒数/节点个数,1)
        long lockWaitTime = calcLockWaitTime(remainTime);

        //调用子类重写方法，locks.size() - minLocksAmount(locks)
        //minLocksAmount(locks) => locks.size()/2 + 1
        //failedLocksLimit = 锁的个数 -（锁的个数/2 + 1）
        //即为Redis节点个数的少数(N/2-1),获取锁容许失败个数的最大阀值为N/2-1，超过这个值，就认定Redlock加锁失败
        int failedLocksLimit = failedLocksLimit();
        List<RLock> acquiredLocks = new ArrayList<>(locks.size());
        for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
            RLock lock = iterator.next();
            boolean lockAcquired;
            try {
                if (waitTime == -1 && leaseTime == -1) {
                    lockAcquired = lock.tryLock();
                } else {
                    //加锁尝试的等待时间为：等待时长的毫秒数 与 Math.max(等待时长的毫秒数/节点个数,1)之间的较小值
                    long awaitTime = Math.min(lockWaitTime, remainTime);
                    //挨个尝试加锁，锁的有效期为等待时长的2倍
                    lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
                }
            } catch (RedisResponseTimeoutException e) {
                //申请锁超时就尝试解锁
                unlockInner(Arrays.asList(lock));
                lockAcquired = false;
            } catch (Exception e) {
                lockAcquired = false;
            }
            
            if (lockAcquired) {
                //加锁成功的节点就放到acquiredLocks这个list中
                acquiredLocks.add(lock);
            } else {
                //加锁失败，须要判断失败的个数是否已经达到了N/2-1个，达到了的话，再来一个失败的，那么这个
                //redlock就加锁失败了，后面的就能够不用再试了
                if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
                    break;
                }

                if (failedLocksLimit == 0) {
                    unlockInner(acquiredLocks);
                    if (waitTime == -1) {
                        return false;
                    }
                    failedLocksLimit = failedLocksLimit();
                    acquiredLocks.clear();
                    // reset iterator
                    while (iterator.hasPrevious()) {
                        iterator.previous();
                    }
                } else {
                    failedLocksLimit--;
                }
            }
            
            if (remainTime != -1) {
                remainTime -= System.currentTimeMillis() - time;
                time = System.currentTimeMillis();
                if (remainTime <= 0) {
                    unlockInner(acquiredLocks);
                    return false;
                }
            }
        }

        //redLock申请成功，为每一个节点上的锁设置过时时间
        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();
            }
        }
        
        return true;
    }
    
复制代码

加锁的代码你们能够经过看注释，根据上文的算法原理解析，应该能够轻松的看懂，接下来再看一下解锁:

protected RFuture<Void> unlockInnerAsync(Collection<RLock> locks, long threadId) {
        if (locks.isEmpty()) {
            return RedissonPromise.newSucceededFuture(null);
        }
        
        RPromise<Void> result = new RedissonPromise<Void>();
        AtomicInteger counter = new AtomicInteger(locks.size());
        for (RLock lock : locks) {
            lock.unlockAsync(threadId).onComplete((res, e) -> {
                if (e != null) {
                    result.tryFailure(e);
                    return;
                }
                //在全部Redis节点上都完成解锁动做后
                if (counter.decrementAndGet() == 0) {
                    result.trySuccess(null);
                }
            });
        }
        return result;
    }
    
复制代码

Redlock虽然是Redis做者踢出的一种实现分布式锁的算法，可是，它也并非一个实现分布式锁的完美算法，若是对Redlock的缺点有兴趣，你们能够看一下Martin Kleppmann批判Redlock的文章(叫Martin的好像都挺牛，Martin Fowler) martin.kleppmann.com/2016/02/08/…

不得不佩服大师，系统各类状况、场景都考虑的很是全面，向大师致敬，咱们在学习知识的同时，也须要本身多思考，多总结。