分布式锁—4.Redisson的联锁和红锁一

大纲 1.Redisson联锁MultiLock概述 2.Redisson联锁MultiLock的加锁与释放锁 3.Redisson红锁RedLock的算法原理 4.Redisson红锁RedLock的源码分析 1.Redisson联锁MultiLock概述 (1)MultiLock的简介 (2)MultiLock的使用 (3)MultiLock的初始化 (1)MultiLock的简介 一.一次性要锁定多个资源的场景 比如锁定一个库存 + 锁定一个订单 + 锁定一个积分，一次性锁定多个资源，这些被锁定的多个资源都不能让其他线程随意修改。然后当前线程一次性更新这些资源后，再逐一释放多个锁。 二.Redisson分布式锁是支持MultiLock机制的 可以将多个锁合并为一个大锁，对大锁进行统一的加锁申请以及锁释放。即一次性锁定多个资源，再去处理一些事情，然后处理完后再一次性释放所有资源对应的锁。 三.Redisson的RedissonMultiLock Redisson的RedissonMultiLock，可以将多个RLock关联为一个联锁，每个RLock对象实例可以来自于不同的Redisson实例。 (2)MultiLock的使用 //联锁 RedissonClient redissonInstance1 = Redisson.create(config); RedissonClient redissonInstance2 = Redisson.create(config); RedissonClient redissonInstance3 = Redisson.create(config); RLock lock1 = redissonInstance1.getLock(“lock1”); RLock lock2 = redissonInstance2.getLock(“lock2”); RLock lock3 = redissonInstance3.getLock(“lock3”); RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3); //同时加锁：lock1 lock2 lock3，所有的锁都上锁成功才算成功 lock.lock(); //同时释放锁 lock.unlock();

//给lock1、lock2、lock3加锁；如果没有主动释放锁的话，10秒后将会自动释放锁 lock.lock(10, TimeUnit.SECONDS); //加锁等待最多是100秒；加锁成功后如果没有主动释放锁的话，锁会在10秒后自动释放 boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS); lock.unlock(); (3)MultiLock的初始化 public class RedissonMultiLock implements RLock { final List locks = new ArrayList<>(); … public RedissonMultiLock(RLock… locks) { if (locks.length == 0) { throw new IllegalArgumentException(“Lock objects are not defined”); } this.locks.addAll(Arrays.asList(locks)); } … } 2.Redisson联锁MultiLock的加锁与释放锁 (1)联锁的获取(超时时间限制 + 加锁失败数限制) (2)联锁的释放(依次释放锁 + 同步等待锁释放完毕) (1)联锁的获取(超时时间限制 + 加锁失败数限制) 一.RedissonMultiLock的lockInterruptibly()方法每次while循环获取所有锁 加锁的时候，首先会调用RedissonMultiLock的lock()方法，接着会调用RedissonMultiLock的lockInterruptibly()方法。 在RedissonMultiLock的lockInterruptibly()方法中，会先根据联锁的个数来计算获取锁时的等待时间waitTime，然后通过while循环不停地尝试调用tryLock()方法去获取所有的锁。只有获取到所有的锁，while循环才会退出。 二.RedissonMultiLock的tryLock()方法获取锁有超时时间限制 + 加锁失败数限制 在RedissonMultiLock的tryLock()方法中，会依次遍历需要获取的锁，然后调用RLock的tryLock()方法尝试获取每个锁。比如调用可重入锁RedissonLock.tryLock()方法来尝试获取每个锁。 假设传入的leaseTime = -1，waitTime = 4500，计算出remainTime = 4500。那么传入RedissonLock的tryLock()方法中的参数waitTime为4500，即指定了获取每个锁时的等待超时时间为4500毫秒。如果在4500毫秒内获取不到这个锁，就退出并标记为获取锁失败。此外传入RedissonLock的tryLock()方法中的参数newLeaseTime为-1。表示获取到锁之后，这个锁在多长时间内会自动释放。由于leaseTime是-1，所以newLeaseTime也是-1。所以如果获取到了锁，会启动一个WatchDog在10秒之后去检查锁的持有情况。 在RedissonMultiLock的tryLock()方法的遍历获取锁的for循环中，有两个限制。 限制一：超时时间限制 当获取锁成功时，就将该锁实例添加到一个列表。但不管获取锁成功还是失败，都会递减remainTime。其实remainTime就是获取MultiLock的超时时间，默认每个锁1500毫秒。当发现remainTime小于0，则表示此次获取联锁失败，需释放获取的锁。此时RedissonMultiLock的tryLock()方法便会返回false，继续下一轮尝试。 限制二：加锁失败数限制 当获取锁失败时，先判断是否达到加锁成功的最少数量。如果达到，就可以退出循环，并进行返回。如果还没达到，就对failedLocksLimit递减。当发现failedLocksLimit为0，则表示此次获取联锁失败，需释放获取的锁，同时重置failedLocksLimit的值+清空acquiredLocks+复位锁列表的迭代器，为下一次尝试获取全部锁做准备。也就是RedissonMultiLock.tryLock()方法会返回false，继续下一轮尝试。 三.RedissonMultiLock的tryLock()方法获取所有锁失败会继续重试 当RedissonMultiLock的tryLock()方法返回false时，在RedissonMultiLock的lockInterruptibly()方法的while循环中，会再次调用RedissonMultiLock的tryLock()方法来尝试获取联锁。 四.总结 假设要获取的联锁中有n把锁，那么可能会循环很多次去尝试获取这n把锁。默认情况下，每次获取这n把锁的时候，会有一个超时时间为1500n毫秒。也就是说，如果第一次获取这n把锁时，在1500n毫秒内无法获取这n把锁。那么就会继续调用tryLock方法进行下一次尝试，重新再来获取这n把锁。直到某一次成功在1500*n毫秒内获取到这n把锁，那么就会退出循环。 public class RedissonMultiLock implements RLock { final List locks = new ArrayList<>(); public RedissonMultiLock(RLock… locks) { … this.locks.addAll(Arrays.asList(locks)); } @Override public void lock() { … lockInterruptibly(); … } @Override public void lockInterruptibly() throws InterruptedException { lockInterruptibly(-1, null); } @Override public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException { //根据联锁的个数来计算获取锁时的等待时间waitTime //此时MutiLock中有3个锁，leaseTime=-1，baseWaitTime=4500，waitTime=4500 long baseWaitTime = locks.size() * 1500; long waitTime = -1; if (leaseTime == -1) { //传入的leaseTime为-1，将baseWaitTime赋值给waitTime waitTime = baseWaitTime; } else { … } //不停地尝试去获取所有的锁 while (true) { //只有获取到所有的锁，while循环才会退出 if (tryLock(waitTime, leaseTime, TimeUnit.MILLISECONDS)) { return; } } } @Override public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException { //此时传入的leaseTime=-1，waitTime=4500，计算出remainTime=4500 long newLeaseTime = -1; … //time=当前时间 long time = System.currentTimeMillis(); long remainTime = -1; if (waitTime != -1) { //remainTime=4500 remainTime = unit.toMillis(waitTime); } //RedissonRedLock会重载calcLockWaitTime()方法，缩短了获取每个小锁的超时时间 //比如RedissonRedLock.calcLockWaitTime()方法返回1500 //RedissonMultiLock.calcLockWaitTime()方法返回4500 long lockWaitTime = calcLockWaitTime(remainTime); //RedissonRedLock会重载failedLocksLimit()方法，返回可以允许最多有多少个锁获取失败 //比如RedissonMultiLock.failedLocksLimit()方法返回0，表示不允许存在某个锁获取失败 int failedLocksLimit = failedLocksLimit(); //acquiredLocks用来保存已获取到的锁 List acquiredLocks = new ArrayList<>(locks.size()); //依次遍历要获取的锁 for (ListIterator iterator = locks.listIterator(); iterator.hasNext();) { RLock lock = iterator.next(); boolean lockAcquired; … if (waitTime == -1 && leaseTime == -1) { lockAcquired = lock.tryLock(); } else { //awaitTime=4500 long awaitTime = Math.min(lockWaitTime, remainTime); //获取锁的核心方法RLock.tryLock()，比如RedissonLock.tryLock()方法 //如果在awaitTime=4500毫秒内获取不到这个锁，就退出并标记为获取锁失败 lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS); } … if (lockAcquired) { //成功获取锁，就将锁实例添加到acquiredLocks acquiredLocks.add(lock); } else { if (locks.size() - acquiredLocks.size() == failedLocksLimit()) { break; } //获取锁失败，就对failedLocksLimit递减，直到failedLocksLimit为0就返回false if (failedLocksLimit == 0) { //此次获取联锁失败，需释放获取的锁 unlockInner(acquiredLocks); if (waitTime == -1) { return false; } //重置failedLocksLimit的值，为下一次尝试获取全部锁做准备 failedLocksLimit = failedLocksLimit(); //清空acquiredLocks，为下一次尝试获取全部锁做准备 acquiredLocks.clear(); //复位锁列表的迭代器 while (iterator.hasPrevious()) { iterator.previous(); } } else { //递减failedLocksLimit failedLocksLimit–; } } //递减remainTime，如果remainTime小于0，表示获取联锁失败 if (remainTime != -1) { remainTime -= System.currentTimeMillis() - time; time = System.currentTimeMillis(); //如果发现remainTime小于0，则表示此次获取联锁失败 if (remainTime <= 0) { unlockInner(acquiredLocks); return false; } } } if (leaseTime != -1) { acquiredLocks.stream() .map(l -> (RedissonLock) l) .map(l -> l.expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS)) .forEach(f -> f.toCompletableFuture().join()); } return true; } … } (2)联锁的释放(依次释放锁 + 同步等待锁释放完毕) 释放锁就是依次调用每个锁的释放逻辑，同步等待每个锁释放完毕才返回。 public class RedissonMultiLock implements RLock { … @Override public void unlock() { List> futures = new ArrayList<>(locks.size()); //依次调用每个锁的释放逻辑 for (RLock lock : locks) { futures.add(lock.unlockAsync()); } for (RFuture future : futures) { //同步等待每个锁释放完毕 future.toCompletableFuture().join(); } } … }