分布式锁加锁失败后的等待逻辑是如何实现的
分布式锁加锁失败后的等待逻辑是如何实现的
面试指北 · 蔚来
在 Redisson 的源码中，监听锁释放的逻辑主要由 LockPubSub 类负责，它利用了 Redis 的 Pub/Sub 机制 来监听锁的释放事件。
锁释放监听的实现过程
- 锁释放时，向 Redis 频道发布消息
-
当某个线程释放锁时，Redisson 的
unlock方法会向对应的 Redis 频道（redisson_lock__channel:<锁名称>）发布一条解锁消息。 -
发布消息的逻辑在
unlockInnerAsync方法中：
java redis.call(ARGV[4], KEYS[2], ARGV[1]);ARGV[4]是 Redis 发布命令PUBLISH。KEYS[2]是锁释放通知的频道名称，例如redisson_lock__channel:<锁名称>。ARGV[1]是解锁消息的内容。
- 订阅锁释放事件
-
当某线程尝试获取锁但失败时，会通过
subscribe方法订阅 Redis 的解锁频道。 -
subscribe方法的实现：
java protected CompletableFuture<RedissonLockEntry> subscribe(long threadId) { return pubSub.subscribe(getEntryName(), getChannelName()); }getChannelName()：生成锁释放的频道名称，类似于redisson_lock__channel:<锁名称>。pubSub.subscribe(...)：订阅指定频道，并返回一个CompletableFuture，表示订阅完成的异步状态。
- 监听消息并唤醒等待线程
-
当锁释放时，Redis 会向频道发布消息，订阅的线程会收到该消息。
-
LockPubSub会调用注册的回调函数，通知所有等待的线程：
java @Override protected void onMessage(RedissonLockEntry value) { // 唤醒所有等待线程 value.getLatch().release(); value.notifyListeners(); }value.getLatch().release()：释放线程等待的Latch。value.notifyListeners()：通知线程重新尝试获取锁。
- 线程唤醒后重新尝试获取锁
-
被唤醒的线程会调用
tryAcquire方法，重新尝试获取锁：
java while (true) { Long ttl = tryAcquire(...); if (ttl == null) { // 锁成功获取 break; } }
锁释放监听的关键点
1. 频道名称
频道名称是 redisson_lock__channel:<锁名称>，由以下方法生成：
String getChannelName() {
return prefixName("redisson_lock__channel", getRawName());
}
2. 发布消息
当锁被释放时，unlock 方法会通过 Redis 的 PUBLISH 命令向频道发布消息：
redis.call('PUBLISH', KEYS[2], ARGV[1]);
3. 订阅消息
锁的订阅通过 LockPubSub 类完成，订阅的逻辑如下：
pubSub.subscribe(getEntryName(), getChannelName());
getChannelName()是要订阅的频道名称。subscribe方法会注册一个监听器，当消息到达时会触发回调。
4. 消息到达时唤醒等待线程
- 当消息到达时，
onMessage
方法被调用：
java
protected void onMessage(RedissonLockEntry value) {
value.getLatch().release(); // 唤醒等待线程
value.notifyListeners(); // 通知监听器
}
完整流程
- 加锁失败
：
- 线程订阅锁的释放频道并进入等待状态（通过
Latch阻塞）。
- 锁释放
：
- 持有锁的线程调用
unlock方法，Redis 发布解锁消息。
- 收到消息
：
- 订阅频道的线程通过
LockPubSub接收到解锁通知。
- 唤醒线程
：
onMessage方法释放Latch，唤醒等待的线程。
- 重新尝试加锁
：
- 被唤醒的线程调用
tryAcquire方法重新尝试获取锁。
代码示例：关键类的简化逻辑
发布消息
-- Redis 脚本解锁时发布消息
if (redis.call('del', KEYS[1]) == 1) then
redis.call('PUBLISH', KEYS[2], ARGV[1]); -- 发布锁释放消息
return 1;
end
订阅逻辑
protected CompletableFuture<RedissonLockEntry> subscribe(long threadId) {
return pubSub.subscribe(getEntryName(), getChannelName());
}
接收消息并唤醒
@Override
protected void onMessage(RedissonLockEntry value) {
value.getLatch().release(); // 唤醒等待线程
value.notifyListeners(); // 通知监听器
}
核心代码
lock
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return;
}
CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
pubSub.timeout(future);
RedissonLockEntry entry;
if (interruptibly) {
entry = commandExecutor.getInterrupted(future);
} else {
entry = commandExecutor.get(future);
}
try {
while (true) {
ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
break;
}
// waiting for message
if (ttl >= 0) {
try {
entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
if (interruptibly) {
throw e;
}
entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
}
} else {
if (interruptibly) {
entry.getLatch().acquire();
} else {
entry.getLatch().acquireUninterruptibly();
}
}
}
} finally {
unsubscribe(entry, threadId);
}
// get(lockAsync(leaseTime, unit));
}
解锁
protected RFuture<Boolean> unlockInnerAsync(long threadId, String requestId, int timeout) {
return evalWriteSyncedNoRetryAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"local val = redis.call('get', KEYS[3]); " +
"if val ~= false then " +
"return tonumber(val);" +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; " +
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
"if (counter > 0) then " +
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"redis.call('set', KEYS[3], 0, 'px', ARGV[5]); " +
"return 0; " +
"else " +
"redis.call('del', KEYS[1]); " +
"redis.call(ARGV[4], KEYS[2], ARGV[1]); " +
"redis.call('set', KEYS[3], 1, 'px', ARGV[5]); " +
"return 1; " +
"end; ",
Arrays.asList(getRawName(), getChannelName(), getUnlockLatchName(requestId)),
LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime,
getLockName(threadId), getSubscribeService().getPublishCommand(), timeout);
}
总结
Redisson 的锁释放监听机制完全依赖于 Redis 的 Pub/Sub 功能：
- 释放锁时：通过
PUBLISH向频道发布解锁消息。 - 订阅锁释放：加锁失败的线程订阅频道并进入等待。
- 消息通知：当锁释放时，订阅者通过
Latch被唤醒，重新尝试获取锁。
这种机制保证了加锁失败的线程不会频繁轮询 Redis，极大地优化了资源消耗，同时提高了锁的公平性。