简介
说明
本文介绍Redis的Redisson客户端的分布式锁的原理。
项目中经常用到分布式锁，而Redis是分布式锁最常用的一种方案。
分布式锁就要考虑锁的续期、释放、可重入、互斥等问题。Redisson这个客户端是目前最完美的一种方案，它在内部可以对锁进行自动续期，程序执行结束、发生异常或者整个应用挂掉都可以释放锁，可重入和互斥也都处理的很好。
有Redisson了，我们没必要自己手写分布式锁了，手写的分布式锁不如Redisson考虑的全面的。
官网
REDIS distlock — Redis中国用户组（CRUG）
Redisson分布式锁方案优点
- Redisson 通过 Watch Dog（看门狗） 机制很好的解决了锁的续期问题。
- 通过 Redisson 实现分布式可重入锁，比原生的SET mylock userId NX PX milliseconds + lua 实现的效果更好。
- 在进程等待申请锁的实现上也做了一些优化，减少了无效的锁申请，提升了资源的利用率。
分布式锁要注意的问题
- 安全属性（Safety property）: 独享（相互排斥）
- 在任意一个时刻，只有一个客户端持有锁。
- 活性A(Liveness property A): 无死锁
- 即便持有锁的客户端崩溃（crashed)或者网络被分裂（gets partitioned)，锁仍然可以被获取。
- 活性B(Liveness property B): 容错
- 只要大部分Redis节点都活着，客户端就可以获取和释放锁
为什么Redisson不用setnx实现分布式锁？
Redisson没有使用setnx命令实现分布式锁，因为虽然setnx命令能够实现分布式锁，但存在以下几个问题：
- 锁过期时间不能自动续约。
- 使用setnx命令实现分布式锁时，如果获取锁的客户端执行时间过长，导致锁过期，其它客户端就有可能获取到这个锁。因此需要加入自动续约机制，在锁的持有者自身没有释放锁的情况下，对锁进行续约以保证该锁持续生效。
- 不支持可重入。
- 如果某个线程已经持有了一个锁，再次对这个锁进行加锁时，setnx命令会认为这个键已经存在，无法再次进行加锁。而支持可重入的锁允许同一线程多次加锁，且要求解锁次数与加锁次数相等。
- 不支持锁的释放。
- setnx命令只能通过手动设置过期时间或者等待过期时间释放锁。如果某个线程异常退出或者未能及时释放锁，就有可能导致死锁的发生。而支持释放锁的锁允许设置锁的自动释放时间或者手动释放锁。
Redisson提供的分布式锁可以解决以上三个问题，并提供了更为完善的分布式锁功能，使得使用Redisson实现分布式锁更加方便和稳定。例如：使用Lua脚本来保证原子性，使用Redis的watch机制来实现分布式锁的释放，使用watchdog机制实现分布式锁的续期。
Redisson分布式锁方案缺点
有个别观点说使用 Watch Dog 机制开启一个定时线程去不断延长锁的时间对系统有所损耗（这里只是网络上的一种说法，博主查了很多资料并且结合实际生产并不认为有很大系统损耗，这个仅供大家参考）。
原理分析
用法
RLock lock = redisson.getLock("myLock");
lock.lock();
try {
// do sth.
} finally {
lock.unlock();
}
加锁
概述
Redisson使用Redis的发布订阅机制来加锁。流程是：获取锁，若获取不成功则订阅释放锁的消息，在收到释放锁的消息前阻塞，收到释放锁的消息后再去循环获取锁。
代码调用流程
lock() //org.redisson.RedissonLock.java#lock
lock(-1, null, false)
tryAcquire(-1, leaseTime, unit, threadId)
tryAcquireAsync(waitTime, leaseTime, unit, threadId)
tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); //这是最重要的方法
lock(-1, null, false)
总结：获取锁，若获取不成功则订阅释放锁的消息，在收到释放锁的消息前阻塞，收到释放锁的消息后再去循环获取锁。
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
long threadId = Thread.currentThread().getId();
// 获取锁
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// 获取成功
if (ttl == null) {
return;
}
// 异步订阅redis channel
RFuture<RedissonLockEntry> future = subscribe(threadId);
if (interruptibly) {
commandExecutor.syncSubscriptionInterrupted(future);
} else {
commandExecutor.syncSubscription(future);
}
try {
while (true) {
ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
break;
}
// waiting for message
if (ttl >= 0) {
try {
future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
if (interruptibly) {
throw e;
}
future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
}
} else {
if (interruptibly) {
future.getNow().getLatch().acquire();
} else {
future.getNow().getLatch().acquireUninterruptibly();
}
}
}
} finally {
// 取消订阅
unsubscribe(future, threadId);
}
// get(lockAsync(leaseTime, unit));
}
tryAcquire(leaseTime, unit, threadId)
private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
return get(tryAcquireAsync(leaseTime, unit, threadId));// 通过异步获取锁，但get(future)实现同步
}
tryAcquireAsync(waitTime, leaseTime, unit, threadId)
总结：用到了Netty的Future-listen模型：给Future一个Promise。
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
if (leaseTime != -1) { //1 如果设置了超时时间，直接调用 tryLockInnerAsync
return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}
//2 如果leaseTime==-1，则默认超时时间为30s
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(LOCK_EXPIRATION_INTERVAL_SECONDS, TimeUnit.SECONDS, threadId, RedisCommands.EVAL_LONG);
//3 监听Future，获取Future返回值ttlRemaining(剩余超时时间)，获取锁成功，但是ttlRemaining，则刷新过期时间
ttlRemainingFuture.addListener(new FutureListener<Long>() {
@Override
public void operationComplete(Future<Long> future) throws Exception {
if (!future.isSuccess()) {
return;
}
Long ttlRemaining = future.getNow();
// lock acquired
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
tryLockInnerAsync
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
return commandExecutor.evalWriteAsync(
getName(),
LongCodec.INSTANCE,
command,
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);",
Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}
脚本入参
| 参数 | 示例值 | 含义 |
| KEY个数 | 1 | KEY个数。为了后面可重入做的计数统计。 |
| KEYS[1] | “myLock” | 加锁的 key 名字 |
| ARGV[1] | 60000 | 持有锁的有效时间：毫秒。默认 30 秒 |
| ARGV[2] | 285475da-9152-4c83-822a-67ee2f116a79:52 | 唯一标识：Redisson客户端ID(UUID)+线程ID |
看一下在 Redis 中的存储结构：
127.0.0.1:6379> HGETALL myLock 1) "285475da-9152-4c83-822a-67ee2f116a79:52" 2) "1"
脚本解读
-- 若锁不存在：新增锁，设置锁重入计数为1，设置锁过期时间，返回
if (redis.call('exists', KEYS[1]) == 0) then
redis.call('hset', KEYS[1], ARGV[2], 1);
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
-- 若锁存在，且唯一标识也匹配：表明当前加锁请求为锁重入请求，故锁重入计数+1，再次设置锁过期时间，返回
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
redis.call('hincrby', KEYS[1], ARGV[2], 1);
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
-- 若锁存在，但唯一标识不匹配：表明锁是被其他线程占用，当前线程无权解他人的锁，直接返回锁剩余过期时间
return redis.call('pttl', KEYS[1]);
互斥
概述
如果客户端 2 来尝试加锁，会如何呢？
首先，第一个 if 判断会执行exists myLock，发现 myLock 这个锁 key 已经存在了。接着第二个 if 判断，判断一下，myLock 锁 key 的 hash 数据结构中，是否包含客户端 2 的 ID，这里明显不是，因为那里包含的是客户端 1 的 ID。所以，客户端 2 会执行：
return redis.call('pttl', KEYS[1]);
返回的一个数字，这个数字代表了 myLock 这个锁 key 的剩余生存时间。
看一下 Redissson tryLock 的主流程：
@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
long time = unit.toMillis(waitTime);
long current = System.currentTimeMillis();
long threadId = Thread.currentThread().getId();
// 1.尝试获取锁
Long ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return true;
}
// 申请锁的耗时如果大于等于最大等待时间，则申请锁失败.
time -= System.currentTimeMillis() - current;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
current = System.currentTimeMillis();
/**
* 2.订阅锁释放事件，并通过 await 方法阻塞等待锁释放，有效的解决了无效的锁申请浪费资源的问题：
* 基于信息量，当锁被其它资源占用时，当前线程通过 Redis 的 channel 订阅锁的释放事件，
* 一旦锁释放会发消息通知待等待的线程进行竞争.
*
* 当 this.await 返回 false，说明等待时间已经超出获取锁最大等待时间，取消订阅并返回获取锁失败.
* 当 this.await 返回 true，进入循环尝试获取锁.
*/
RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
// await 方法内部是用 CountDownLatch 来实现阻塞，获取 subscribe 异步执行的结果（应用了 Netty 的 Future）
if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
if (!subscribeFuture.cancel(false)) {
subscribeFuture.onComplete((res, e) -> {
if (e == null) {
unsubscribe(subscribeFuture, threadId);
}
});
}
acquireFailed(threadId);
return false;
}
try {
// 计算获取锁的总耗时，如果大于等于最大等待时间，则获取锁失败.
time -= System.currentTimeMillis() - current;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
/**
* 3.收到锁释放的信号后，在最大等待时间之内，循环一次接着一次的尝试获取锁
* 获取锁成功，则立马返回 true，
* 若在最大等待时间之内还没获取到锁，则认为获取锁失败，返回 false 结束循环
*/
while (true) {
long currentTime = System.currentTimeMillis();
// 4.再次尝试获取锁
ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return true;
}
// 5.超过最大等待时间则返回 false 结束循环，获取锁失败
time -= System.currentTimeMillis() - currentTime;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
/**
* 6.阻塞等待锁（通过信号量(共享锁)阻塞,等待解锁消息）：
*/
currentTime = System.currentTimeMillis();
if (ttl >= 0 && ttl < time) {
//如果剩余时间(ttl)小于wait time ,就在 ttl 时间内，从Entry的信号量获取
//一个许可(除非被中断或者一直没有可用的许可)。
getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
//则就在wait time 时间范围内等待可以通过信号量
getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
}
// 更新剩余的等待时间(最大等待时间-已经消耗的阻塞时间)
time -= System.currentTimeMillis() - currentTime;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
}
} finally {
// 7.无论是否获得锁,都要取消订阅解锁消息
unsubscribe(subscribeFuture, threadId);
}
// return get(tryLockAsync(waitTime, leaseTime, unit));
}
流程分析：
- 尝试获取锁。若返回 null 则说明加锁成功；若返回一个数值，则说明已经存在该锁，ttl 为锁的剩余存活时间。
- 如果此时客户端 2 进程获取锁失败，那么使用客户端 2 的线程 id（其实本质上就是进程 id）通过 Redis 的 channel 订阅锁释放的事件，。如果等待的过程中一直未等到锁的释放事件通知，当超过最大等待时间则获取锁失败，返回 false，也就是第 39 行代码。如果等到了锁的释放事件的通知，则开始进入一个不断重试获取锁的循环。
- 循环中每次都先试着获取锁，并得到已存在的锁的剩余存活时间。如果在重试中拿到了锁，则直接返回。如果锁当前还是被占用的，那么等待释放锁的消息，具体实现使用了 JDK 的信号量 Semaphore 来阻塞线程，当锁释放并发布释放锁的消息后，信号量的release()方法会被调用，此时被信号量阻塞的等待队列中的一个线程就可以继续尝试获取锁了。
特别注意：
以上过程存在一个细节，这里有必要说明一下，也是分布式锁的一个关键点：当锁正在被占用时，等待获取锁的进程并不是通过一个 while(true) 死循环去获取锁，而是利用了 Redis 的发布订阅机制，通过 await 方法阻塞等待锁的进程，有效的解决了无效的锁申请浪费资源的问题。
续期
概述
客户端 1 加锁的锁 key 默认生存时间才 30 秒，如果超过了 30 秒，客户端 1 还想一直持有这把锁，怎么办呢？
Redisson 提供了一个续期机制， 只要客户端 1 一旦加锁成功，就会启动一个 Watch Dog。可以这样来使用看门狗（leaseTime不设置，或者设置为-1）
lock.tryLock() lock.tryLock(xxx, -1, xxx)
源码
org.redisson.RedissonLock#tryAcquireAsync
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
if (leaseTime != -1) {
return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// lock acquired
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId);
}
});
return ttlRemainingFuture;
}
注意：从以上源码我们看到 leaseTime 必须是 -1 才会开启 Watch Dog 机制，也就是如果你想开启 Watch Dog 机制必须使用默认的加锁时间为 30s。如果你自己自定义时间，超过这个时间，锁就会自己释放，并不会延长。
private void scheduleExpirationRenewal(long threadId) {
ExpirationEntry entry = new ExpirationEntry();
ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
if (oldEntry != null) {
oldEntry.addThreadId(threadId);
} else {
entry.addThreadId(threadId);
renewExpiration();
}
}
protected RFuture<Boolean> renewExpirationAsync(long threadId) {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return 1; " +
"end; " +
"return 0;",
Collections.<Object>singletonList(getName()),
internalLockLeaseTime, getLockName(threadId));
}
Watch Dog 机制其实就是一个后台定时任务线程。获取锁成功之后，会将持有锁的线程放入到一个 RedissonLock.EXPIRATION_RENEWAL_MAP里面，然后每隔 10 秒（internalLockLeaseTime / 3） 检查一下，如果客户端 1 还持有锁 key（判断客户端是否还持有 key，其实就是遍历 EXPIRATION_RENEWAL_MAP 里面线程 id 然后根据线程 id 去 Redis 中查，如果存在就会延长 key 的时间），那么就会不断的延长锁 key 的生存时间。
注意：这里有一个细节：如果服务宕机了，Watch Dog 机制的线程也就没有了，此时就不会延长 key 的过期时间，到了 30s 之后就会自动过期了，其他线程就可以获取到锁。
可重入
Redisson 也是支持可重入锁的，比如下面这种代码：
@Override
public void lock() {
RLock lock = redissonSingle.getLock("myLock");
try {
lock.lock();
// 执行业务
doBusiness();
lock.lock();
} catch (Exception e) {
e.printStackTrace();
} finally {
// 释放锁
lock.unlock();
lock.unlock();
logger.info("任务执行完毕, 释放锁!");
}
}
我们再分析一下加锁那段 lua 代码：
if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);"
第一个 if 判断肯定不成立，exists myLock 会显示锁 key 已经存在。第二个 if 判断会成立，因为 myLock 的 hash 数据结构中包含的那个 ID 即客户端 1 的 ID，此时就会执行可重入加锁的逻辑，使用：hincrby myLock 285475da-9152-4c83-822a-67ee2f116a79:52 1对客户端 1 的加锁次数加 1。此时 myLock 数据结构变为下面这样：
127.0.0.1:6379> HGETALL myLock 1) "285475da-9152-4c83-822a-67ee2f116a79:52" 2) "2"
到这里，小伙伴本就都明白了 hash 结构的 key 是锁的名称，field 是客户端 ID，value 是该客户端加锁的次数。
释放
概述
锁的释放利用了Redis的订阅功能。
释放锁的步骤主要分三步：
- 删除锁（这里注意可重入锁，在上面的脚本中有详细分析）。
- 广播释放锁的消息，通知阻塞等待的进程（向通道名为 redisson_lock__channel publish 一条 UNLOCK_MESSAGE 信息）。
- 取消 Watch Dog 机制，即将 RedissonLock.EXPIRATION_RENEWAL_MAP 里面的线程 id 删除，并且 cancel 掉 Netty 的那个定时任务线程。
源码
执行
lock.unlock()
就可以释放分布式锁。我们来看一下释放锁的流程代码：
@Override
public RFuture<Void> unlockAsync(long threadId) {
RPromise<Void> result = new RedissonPromise<Void>();
// 1. 异步释放锁
RFuture<Boolean> future = unlockInnerAsync(threadId);
// 取消 Watch Dog 机制
future.onComplete((opStatus, e) -> {
cancelExpirationRenewal(threadId);
if (e != null) {
result.tryFailure(e);
return;
}
if (opStatus == null) {
IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
+ id + " thread-id: " + threadId);
result.tryFailure(cause);
return;
}
result.trySuccess(null);
});
return result;
}
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
// 判断锁 key 是否存在
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; " +
// 将该客户端对应的锁的 hash 结构的 value 值递减为 0 后再进行删除
// 然后再向通道名为 redisson_lock__channel publish 一条 UNLOCK_MESSAGE 信息
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
"if (counter > 0) then " +
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
"else " +
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; "+
"end; " +
"return nil;",
Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}
请先 ！