分布式锁—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<RLock> 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把锁的时候，会有一个超时时间为1500*n毫秒。也就是说，如果第一次获取这n把锁时，在1500*n毫秒内无法获取这n把锁。那么就会继续调用tryLock方法进行下一次尝试，重新再来获取这n把锁。直到某一次成功在1500*n毫秒内获取到这n把锁，那么就会退出循环。
public class RedissonMultiLock implements RLock {
final List<RLock> 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<RLock> acquiredLocks = new ArrayList<>(locks.size());
//依次遍历要获取的锁
for (ListIterator<RLock> 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<RFuture<Void>> futures = new ArrayList<>(locks.size());
//依次调用每个锁的释放逻辑
for (RLock lock : locks) {
futures.add(lock.unlockAsync());
}
for (RFuture<Void> future : futures) {
//同步等待每个锁释放完毕
future.toCompletableFuture().join();
}
}
...
}
3.Redisson红锁RedLock的算法原理
(1)RedLock算法的具体流程
(2)RedLock算法的四个要点总结
(1)RedLock算法的具体流程
步骤一：客户端先获取当前时间戳T1。
步骤二：客户端依次向这5个节点发起加锁请求，且每个请求都会设置超时时间。超时时间是毫秒级的，要远小于锁的有效时间，而且一般是几十毫秒。如果某一个节点加锁失败，包括网络超时、锁被其它线程持有等各种情况，那么就立即向下一个Redis节点申请加锁。
步骤三：如果客户端从3个以上(过半)节点加锁成功，则再次获取当前时间戳T2。如果T2 - T1 < 锁的过期时间，则认为客户端加锁成功，否则加锁失败。
步骤四：如果加锁失败，要向全部节点发起释放锁的请求。如果加锁成功，则去操作共享资源。
(2)RedLock算法的四个要点总结
一.客户端在多个Redis节点上申请加锁
二.必须保证大多数节点加锁成功
三.大多数节点加锁的总耗时 < 锁设置的过期时间
四.释放锁时要向全部节点发起释放锁的请求
4.Redisson红锁RedLock的源码分析
(1)RedLock的使用简介
(2)RedLock的实现
(3)RedissonRedLock的源码总结
(1)RedLock的使用简介
//红锁
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");
RedissonRedLock lock = new RedissonRedLock(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();
(2)RedLock的实现
RedissonRedLock锁的实现非常简单，因为RedissonRedLock是RedissonMultiLock的一个子类，所以RedLock的算法是依赖MultiLock的机制来实现的。
RedissonRedLock主要会通过方法的重载，来改变RedissonMultiLock中的几个特殊行为。
一.RedissonRedLock重载了RedissonMultiLock的failedLocksLimit()方法
failedLocksLimit()方法会返回允许最多有多少个锁获取失败。其中failedLocksLimit()方法会调用minLocksAmount()方法，而minLocksAmount()方法便会返回加锁成功的最少数量，即过半数。锁的总数减去加锁成功的最少数量，便是允许最多有多少个锁获取失败。
RedissonMultiLock的failedLocksLimit()方法是返回0的，即RedissonMultiLock是不允许存在某个锁获取失败。
具体的处理就是在RedissonMultiLock的tryLock()方法中，当获取锁失败时，先判断是否达到加锁成功的最少数量。如果达到，就可以退出循环，并进行返回。如果还没达到，就对failedLocksLimit递减。当发现failedLocksLimit为0，则表示此次获取联锁失败，需要释放获取的锁，同时重置failedLocksLimit的值 + 清空acquiredLocks + 复位锁列表的迭代器，为下一次尝试获取全部锁做准备。也就是RedissonMultiLock的tryLock()方法会返回false，继续下一轮尝试。
二.RedissonRedLock重载了RedissonMultiLock的calcLockWaitTime()方法
calcLockWaitTime()方法会返回对每个lock进行加锁时的超时时间。例如当waitTime = 4500毫秒、remainTime = 4500毫秒时：RedissonMultiLock的calcLockWaitTime()方法会返回4500，RedissonRedLock的calcLockWaitTime()方法会返回1500。
RedissonMultiLock中对每个lock尝试加锁的超时时间为4500毫秒，RedissonRedLock中对每个lock尝试加锁的超时时间为1500毫秒。如果在超时时间内没获取到锁，那么就认为对lock的加锁失败。
public class RedissonRedLock extends RedissonMultiLock {
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);
}
}
public class RedissonMultiLock implements RLock {
...
@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<RLock> acquiredLocks = new ArrayList<>(locks.size());
//依次遍历要获取的锁
for (ListIterator<RLock> 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;
}
...
}
public class RedissonMultiLock implements RLock {
...
protected int failedLocksLimit() {
return 0;
}
protected long calcLockWaitTime(long remainTime) {
return remainTime;
}
@Override
public void unlock() {
List<RFuture<Void>> futures = new ArrayList<>(locks.size());
for (RLock lock : locks) {
futures.add(lock.unlockAsync());
}
for (RFuture<Void> future : futures) {
future.toCompletableFuture().join();
}
}
protected void unlockInner(Collection<RLock> locks) {
locks.stream().map(RLockAsync::unlockAsync)
.forEach(f -> {
f.toCompletableFuture().join();
}
);
}
...
}
(3)RedissonRedLock的源码总结
针对多个lock进行加锁，每个lock都有一个1500毫秒的加锁超时时间。
如果在1500*n毫秒内，成功对n / 2 + 1个lock加锁成功了。那么就可以认为这个RedLock加锁成功，不要求所有的lock都加锁成功。
问题：RedLock本应该是一个锁，只不过是在不同的Master节点上进行加锁。但是Redisson的RedLock实现中却通过合并多个小lock来实现，这是否与RedLock的设计不一致了？
当使用Redis Cluster时，其实是一样的。假设有3个Master实例，那么就使用lock1、lock2、lock3三个key去加锁。这3个锁key会按照CRC16得出Hash值然后再取模分布到这3个Master节点，效果等同于让各个Master节点使用名为lock的key进行加锁。
详细介绍后端技术栈的基础内容，包括但不限于：MySQL原理和优化、Redis原理和应用、JVM和G1原理和优化、RocketMQ原理应用及源码、Kafka原理应用及源码、ElasticSearch原理应用及源码、JUC源码、Netty源码、zk源码、Dubbo源码、Spring源码、Spring Boot源码、SCA源码、分布式锁源码、分布式事务、分库分表和TiDB、大型商品系统、大型订单系统等
