Redisson 源码分析 —— 可重入分布式锁 ReentrantLock 芋道源码 2020年12月05日 redis 预计阅读 56 分钟 原文 > `转载`于【[芋道源码](http://svip.iocoder.cn/)】 ## 1. 概述 在 Redisson 中，提供了 8 种分布锁的实现，具体我们可以在 [《Redisson 文档 —— 分布式锁和同步器》](https://github.com/redisson/redisson/wiki/8.-分布式锁和同步器) 中看到。绝大数情况下，我们使用可重入锁（Reentrant Lock）就够了，对应到就是 [`org.redisson.RedissonLock`](https://github.com/YunaiV/redisson/blob/master/redisson/src/main/java/org/redisson/RedissonLock.java) 类，具体的使用示例可以看看 [《芋道 Spring Boot Redis 入门》](http://www.iocoder.cn/Spring-Boot/Redis/?vip) 的「6.2 Redis 分布式锁」小节 。 在 [《精尽 Redis 面试题》](http://svip.iocoder.cn/Redis/Interview/?self) 的问题中，我们在聊到**“如何使用 Redis 实现分布式锁？”**这个题目中，提到了需要考虑的 7 个方面，这里我们再来重复看下： - 1、正确的获得锁 > set 指令附带 nx 参数，保证有且只有一个进程获得到。 - 2、正确的释放锁 > 使用 Lua 脚本，比对锁持有的是不是自己。如果是，则进行删除来释放。 - 3、超时的自动释放锁 > set 指令附带 expire 参数，通过过期机制来实现超时释放。 - 4、未获得到锁的等待机制 > sleep 或者基于 Redis 的订阅 Pub/Sub 机制。 > > 一些业务场景，可能需要支持获得不到锁，直接返回 false ，不等待。 - 5、【可选】锁的重入性 > 通过 ThreadLocal 记录是第几次获得相同的锁。 > > 1）有且第一次计数为 1 && 获得锁时，才向 Redis 发起获得锁的操作。 > 2）有且计数为 0 && 释放锁时，才向 Redis 发起释放锁的操作。 - 6、锁超时的处理 > 一般情况下，可以考虑告警 + 后台线程自动续锁的超时时间。通过这样的机制，保证有且仅有一个线程，正在持有锁。 - 7、Redis 分布式锁丢失问题 > 具体看「方案二：Redlock」。 RedissonLock 实现了前 6 点，而第 7 点需要通过 [`org.redisson.RedissonRedLock`](https://github.com/YunaiV/redisson/blob/master/redisson/src/main/java/org/redisson/RedissonRedLock.java) 来实现，这个话题，我们后面在聊。 在开始阅读源码之前，胖友可以先看看艿艿画的 Redisson 实现分布式锁的整体流程图，以便更好的阅读源码。[猛击传送门](https://www.processon.com/view/link/5f4c871d079129356ec6f4d7) ## 2. 整体一览 我们来看看 Redisson 锁相关的整体类图，如下： [](http://www.iocoder.cn/images/Redis/2019_10_04/01.png)RLock 接口 - [`org.redisson.api.RLockAsync`](https://github.com/YunaiV/redisson/blob/master/redisson/src/main/java/org/redisson/api/RLockAsync.java) ，定义了异步操作的接口。 - [`org.redisson.api.RLock`](https://github.com/YunaiV/redisson/blob/master/redisson/src/main/java/org/redisson/api/RLock.java) ，继承 RLockAsync 的基础上，定义了同步操作的接口。比较有意思的是，RLock 同时实现继承 JDK 的 `java.util.concurrent.locks.Lock` 接口，从而符合 Java 的 Lock 的标准。 - 本文的主角 RedissonLock ，实现 RLock 接口，提供可重入的分布式锁实现。 - 其它的 RLock 实现的关系，胖友自己看图哈。 RLockAsync 和 RLock 定义的接口，差别就在于同步和异步，所以我们就只看看 RLock 接口。代码如下： ``` String getName(); // 锁定相关的接口方法，还有部分在 Lock 接口上 void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException; void lock(long leaseTime, TimeUnit unit); boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException; // 解锁相关的接口方法，还有部分在 Lock 接口上 boolean forceUnlock(); // 其它非关键方法 boolean isLocked(); boolean isHeldByThread(long threadId); boolean isHeldByCurrentThread(); int getHoldCount(); long remainTimeToLive(); ``` ## 3. Lua 脚本 在我们看具体的代码实现，我们先来看核心的使用到的 Lua 脚本，方便我们后续更好的理解 RedissonLock 的实现。 ### 3.1 tryLockInnerAsync `#tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command)` 方法，实现**加锁**逻辑，并且支持**可重入性**。代码如下： > FROM [《慢谈 Redis 实现分布式锁 以及 Redisson 源码解析》](https://crazyfzw.github.io/2019/04/15/distributed-locks-with-redis/) > > [](http://static.iocoder.cn/a1ecddbd0ffc0da7e4f85ba6637c72ee)加锁流程图 ``` // RedissonLock.java 1: <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { 2: internalLockLeaseTime = unit.toMillis(leaseTime); 3: 4: return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command, 5: "if (redis.call('exists', KEYS[1]) == 0) then " + // 情况一，当前分布式锁被未被获得 6: "redis.call('hset', KEYS[1], ARGV[2], 1); " + // 写入分布锁被 ARGV[2] 获取到了，设置数量为 1 。 7: "redis.call('pexpire', KEYS[1], ARGV[1]); " + // 设置分布式的过期时间为 ARGV[1] 8: "return nil; " + // 返回 null ，表示成功 9: "end; " + 10: "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + // 情况二，如果当前分布锁已经被 ARGV[2] 持有 11: "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + // 写入持有计数字 + 1 。 12: "redis.call('pexpire', KEYS[1], ARGV[1]); " + // 设置分布式的过期时间为 ARGV[1] 13: "return nil; " + // 返回 null ，表示成功 14: "end; " + 15: "return redis.call('pttl', KEYS[1]);", // 情况三，获取不到分布式锁，则返回锁的过期时间。 16: Collections.<Object>singletonList(getName()), // KEYS[分布式锁名] 17: internalLockLeaseTime, getLockName(threadId)); // ARGV[锁超时时间，获得的锁名] 18: } ``` - `<2>` 处，根据传入的 `leaseTime` + `unit` 参数，设置到 `internalLockLeaseTime` 属性上，表示锁的时长。代码如下： ``` // RedissonLock.java /** * 锁的时长 */ protected long internalLockLeaseTime; public RedissonLock(CommandAsyncExecutor commandExecutor, String name) { // ... 省略其它代码 this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(); } ``` - 默认情况下，`internalLockLeaseTime` 属性，使用 Lock 的 WatchDog 的超时时长 `30 * 1000` 毫秒。默认的值，当且仅当我们未**显示**传入锁的时长时，才有用。例如说，稍后我们会看到的 `#lock()` 等等方法中。 - 有一点，我们要特别注意，`internalLockLeaseTime` 是 RedissonLock 的**成员**变量，并且也未声明 `volatile` 修饰，所以跨线程使用同一个 RedissonLock 对象，可能会存在 `internalLockLeaseTime` 读取不到最新值的情况。 - 还是熟悉的配方，通过 Lua 脚本实现。具体传入的参数，朋友看下第 16 和 17 行的代码，对应的 `KEYS` 和 `ARGV` 。可能有几个值胖友会有点懵逼，我们先来看看。 - `KEYS[1]` ：调用 `#getName()` 方法获得分布式锁的名字。稍后，我们会看到分布式锁在 Redis 使用是以 `KEYS[1]` 分布式锁为 KEY ，VALUE 为 HASH 类型。 - `ARGV[1]` ：锁的时长。 - `ARGV[2]` ：调用 `#getLockName(threadId)` 方法，获得的锁名。该名字，用于表示该分布式锁正在被哪个进程的线程所持有。代码如下： ``` // RedissonLock.java /** * ID ，就是 {@link ConnectionManager#getId()} */ final String id; protected String getLockName(long threadId) { return id + ":" + threadId; } ``` - 可能描述看起来不是很好理解，我们来看一个获取到分布式锁的示例：[](http://www.iocoder.cn/images/Redis/2019_10_04/02.png)分布式锁的示例 - 第 4 至 15 行：Lua 脚本，一共分成 3 种情况，胖友认真仔细看看，艿艿已经添加了完整的注释。 不同于我们在市面上看到的 Redis 通过 SET 命令带上 NX 和 EXPIRE 的方式实现获得分布式锁，RedissonLock 提供重入性，所以需要 Lua 脚本来实现。当然，实际上，也可以通过 ThreadLocal 来实现重入性的技术，胖友可以思考一波，不懂的话星球来给艿艿留言。 ### 3.2 unlockInnerAsync `#unlockInnerAsync(long threadId)` 方法，实现**解锁**逻辑，并且支持**可重入性**。代码如下： > FROM [《慢谈 Redis 实现分布式锁 以及 Redisson 源码解析》](https://crazyfzw.github.io/2019/04/15/distributed-locks-with-redis/) > > [](http://static.iocoder.cn/bbc6ac47cd74c5cd2641486415438cf5)解锁流程图 ``` // RedissonLock.java protected RFuture<Boolean> unlockInnerAsync(long threadId) { return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + // 情况一，分布式锁未被 ARGV[3] 持有，则直接返回 null ，表示解锁失败。 "return nil;" + "end; " + "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " + // 持有锁的数量减 1 。 "if (counter > 0) then " + // 情况二，如果后还有剩余的持有锁数量，则返回 0 ，表示解锁未完成 "redis.call('pexpire', KEYS[1], ARGV[2]); " + // 重新设置过期时间为 ARGV[2] "return 0; " + "else " + // 情况三，不存在剩余的锁数量，则返回 1 ，表示解锁成功 "redis.call('del', KEYS[1]); " + // 删除对应的分布式锁对应的 KEYS[1] "redis.call('publish', KEYS[2], ARGV[1]); " + // 发布解锁事件到 KEYS[2] ，通知气他可能要获取锁的线程 "return 1; "+ "end; " + "return nil;", // 不存在这个情况。 Arrays.<Object>asList(getName(), getChannelName()), // KEYS[分布式锁名, 该分布式锁对应的 Channel 名] LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId)); // ARGV[解锁消息，锁超时时间，获得的锁名] } ``` - 具体传入的参数，朋友看下第 16 和 17 行的代码，对应的 `KEYS` 和 `ARGV` 。 - `KEYS[1]` ：调用 `#getName()` 方法获得分布式锁的名字。 - `KEYS[2]` ：调用 `#getChannelName()` 方法，该分布式锁对应的 Channel 名。因为 RedissonLock 释放锁时，会通过该 Channel 来 Publish 一条消息，通知其它可能在阻塞等待这条消息的客户端。代码如下： ``` // RedissonLock.java String getChannelName() { return prefixName("redisson_lock__channel", getName()); } ``` - 通过 [Redis Pub/Sub](http://redis.cn/topics/pubsub.html) 机制，实现**未获得到锁的等待机制**。 - 每个分布式锁，对应一个其独有的 Channel 。 - `ARGV[1]` ：解锁消息 [`LockPubSub.UNLOCK_MESSAGE`](https://github.com/YunaiV/redisson/blob/master/redisson/src/main/java/org/redisson/pubsub/LockPubSub.java) 。通过收到这条消息，其它等待锁的客户端，会重新发起获得锁的请求。具体的，我们在下文来一起瞅瞅。 - `ARGV[2]` ：锁的时长。 - `ARGV[3]` ：调用 `#getLockName(threadId)` 方法，获得的锁名。 - 第 3 至 15 行：Lua 脚本，还是分成 3 种情况，胖友认真仔细看看，艿艿已经添加了完整的注释。 不同于我们在市面上看到的 Redis 通过 Lua 脚本的方式实现释放分布式锁，一共有 2 点： - 1、要实现**可重入性**，所以只有在计数为 0 时，才会真正释放锁。 - 2、要实现客户端的**等待通知**，所以在释放锁时，Publish 一条释放锁的消息。 ### 3.3 forceUnlockAsync `#forceUnlockAsync()` 方法，实现**强制解锁**逻辑。代码如下： ``` // RedissonLock.java @Override public RFuture<Boolean> forceUnlockAsync() { cancelExpirationRenewal(null); return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('del', KEYS[1]) == 1) then " // 情况一，释放锁成功，则通过 Publish 发布释放锁的消息，并返回 1 + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1 " + "else " // 情况二，释放锁失败，因为不存在这个 KEY ，所以返回 0 + "return 0 " + "end", Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE); } ``` - 比较简单，分成 2 种情况。胖友认真仔细看看，艿艿已经添加了完整的注释。 代码处理的比较细致，Redis DEL 成功，才 PUBLISH 发布释放锁的消息，避免错误通知客户端。 ### 3.4 renewExpirationAsync `#renewExpirationAsync(long threadId)` 方法，实现**续锁**逻辑。可能胖友有点懵逼，至少艿艿看到这段逻辑，完全不知道为何意啊。 我们先来看下 [《Redisson 文档 —— 分布式锁和同步器》](https://github.com/redisson/redisson/wiki/8.-分布式锁和同步器) ，有一段奇怪的说明： ``` RLock lock = redisson.getLock("anyLock"); // 最常见的使用方法 lock.lock(); ``` > 大家都知道，如果负责储存这个分布式锁的Redisson节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改 [Config.lockWatchdogTimeout](https://github.com/redisson/redisson/wiki/2.-配置方法#lockwatchdogtimeout监控锁的看门狗超时单位毫秒) 来另行指定。 在使用 `RedissonLock#lock()` 方法，我们要求**持续**持有锁，直到**手动**释放。但是实际上，我们有一个隐藏条件，如果 Java 进程挂掉时，需要**自动**释放。那么，如果实现 `RedissonLock#lock()` 时，设置过期 Redis 为无限大，或者不过期都不合适。那么 RedissonLock 是怎么实现的呢？RedissonLock 先获得一个 `internalLockLeaseTime` 的分布式锁，然后每 `internalLockLeaseTime / 3` 时间，定时调用 `#renewExpirationAsync(long threadId)` 方法，进行续租。这样，在 Java 进程异常 Crash 掉后，能够保证最多 `internalLockLeaseTime` 时间后，分布式锁自动释放。 略骚略巧妙~不过为了实现这样的功能，RedissonLock 的整体逻辑，又复杂了一丢丢。 下面，还是先让我们看下具体的 `#renewExpirationAsync(long threadId)` 方法的代码，如下： ``` // RedissonLock.java 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 " + // 情况一，如果持有锁，则重新设置过期时间为 ARGV[1] internalLockLeaseTime ，并返回 1 续租成功。 "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return 1; " + "end; " + "return 0;", // 情况二，未吃货有，返回 0 续租失败。 Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); } ``` - 比较简单，分成 2 种情况。胖友认真仔细看看，艿艿已经添加了完整的注释。 ------ 至此，我们看完了 Lua 脚本部分，其实基本也大体知道 RedissonLock 是如何实现加锁、接锁的逻辑。但是，复杂的逻辑，还在下面，胖友请保持好耐心，开启我们的高能时刻。 > 艿艿：T T RedissonLock 好多重载的方法，文章小标题，都不造杂取了。最关键的是，可能胖友会被绕进去。 ## 4. LockPubSub 在开始研究真正的加锁和解锁的调用之前，我们先看看和其相关的客户端订阅解锁消息，从而实现在持有锁的客户端释放锁时，等待锁的客户端能够快速的去调用加锁逻辑。 > 艿艿：😭 整个调用栈太深的，艿艿只好先写它，就当卖了一个关子，哈哈哈。 [`org.redisson.pubsub.LockPubSub`](https://github.com/YunaiV/redisson/blob/master/redisson/src/main/java/org/redisson/pubsub/LockPubSub.java) ，继承 PublishSubscribe 抽象类，实现 Lock 相关消息的订阅。代码如下： ``` // LockPubSub.java public class LockPubSub extends PublishSubscribe<RedissonLockEntry> { /** * 锁释放的消息 */ public static final Long UNLOCK_MESSAGE = 0L; /** * 读锁释放的消息 */ public static final Long READ_UNLOCK_MESSAGE = 1L; public LockPubSub(PublishSubscribeService service) { super(service); } @Override protected RedissonLockEntry createEntry(RPromise<RedissonLockEntry> newPromise) { return new RedissonLockEntry(newPromise); } @Override protected void onMessage(RedissonLockEntry value, Long message) { if (message.equals(UNLOCK_MESSAGE)) { // 回调监听器 Runnable runnableToExecute = value.getListeners().poll(); if (runnableToExecute != null) { runnableToExecute.run(); } // 通过信号量，通知阻塞等待的线程 value.getLatch().release(); } else if (message.equals(READ_UNLOCK_MESSAGE)) { while (true) { Runnable runnableToExecute = value.getListeners().poll(); if (runnableToExecute == null) { break; } runnableToExecute.run(); } value.getLatch().release(value.getLatch().getQueueLength()); } } } ``` - 在 `#createEntry(RPromise<RedissonLockEntry> newPromise)` 方法，会创建 [`org.redisson.RedissonLockEntry`](https://github.com/YunaiV/redisson/blob/master/redisson/src/main/java/org/redisson/RedissonLockEntry.java) 对象。代码如下： ``` // RedissonLockEntry.java public class RedissonLockEntry implements PubSubEntry<RedissonLockEntry> { /** * 计数器 * * 每次发起订阅，则计数器 + 1 * 每次取消订阅，则计数器 - 1 。当减少到 0 时，才正常取消订阅。 */ private int counter; /** * 信号量，用于实现 RedissonLock 阻塞等待的通知 */ private final Semaphore latch; private final RPromise<RedissonLockEntry> promise; /** * 监听器们 */ private final ConcurrentLinkedQueue<Runnable> listeners = new ConcurrentLinkedQueue<Runnable>(); public RedissonLockEntry(RPromise<RedissonLockEntry> promise) { super(); this.latch = new Semaphore(0); this.promise = promise; } @Override public void aquire() { counter++; } @Override public int release() { return --counter; } @Override public RPromise<RedissonLockEntry> getPromise() { return promise; } public void addListener(Runnable listener) { listeners.add(listener); } public boolean removeListener(Runnable listener) { return listeners.remove(listener); } public ConcurrentLinkedQueue<Runnable> getListeners() { return listeners; } public Semaphore getLatch() { return latch; } } ``` - 虽然代码比较多，我们重点来看 `latch` 和 `listeners` 属性。 - `latch` 属性：信号量，用于实现 RedissonLock 阻塞等待的通知。在我们下面看到**同步**加锁的逻辑，会看到通过它来实现阻塞等待。 - `listeners` 属性：监听器，实现订阅到锁的释放消息，从而再次发起获得锁。当然，这里的 Runnable 对象肯定无法体现，具体我们后面看看 `#tryLockAsync(AtomicLong time, long leaseTime, TimeUnit unit, RFuture<RedissonLockEntry> subscribeFuture, RPromise<Boolean> result, long currentThreadId)` 或者 `#lockAsync(long leaseTime, TimeUnit unit, RFuture<RedissonLockEntry> subscribeFuture, RPromise<Void> result, long currentThreadId)` 方法，就可以看到方法内部会创建具体的 Runnable 实现类，实现再次发起获得锁的逻辑。 - 在 `#onMessage(RedissonLockEntry value, Long message)` 方法中，在接收到释放锁的消息后，会执行 `listeners` 的回调，以及 `latch` 的时候放。 当然，单单看 LockPubSub 类，胖友可能会感到懵逼，保持耐心，继续向下。LockPubSub 更多的是实现了锁释放消息的监听，以及回调监听器，释放信号量。**真正的逻辑，还是要看监听器的逻辑，以及 RedissonLock 是怎么实现信号量的**。 另外，在 RedissonLock 中，提供如下几个方法，发起和取消订阅。代码如下： ``` // RedissonLock.java /** * Sub Entry 名字 */ final String entryName; protected final LockPubSub pubSub; public RedissonLock(CommandAsyncExecutor commandExecutor, String name) { // ... 省略其他无关 this.entryName = id + ":" + name; this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub(); } /** * 获得线程对应的 RedissonLockEntry 对象 * * @param threadId 线程编号 * @return RedissonLockEntry 对象 */ protected RedissonLockEntry getEntry(long threadId) { return pubSub.getEntry(getEntryName()); } /** * 异步发起订阅 * * @param threadId 线程编号 * @return RFuture 对象 */ protected RFuture<RedissonLockEntry> subscribe(long threadId) { return pubSub.subscribe(getEntryName(), getChannelName()); } /** * 异步取消订阅 * * @param future RFuture 对象 * @param threadId 线程编号 */ protected void unsubscribe(RFuture<RedissonLockEntry> future, long threadId) { pubSub.unsubscribe(future.getNow(), getEntryName(), getChannelName()); } ``` ## 5. tryLockAsync > 艿艿：重点开始了，打起精神。 `#tryLockAsync(long waitTime, TimeUnit unit)` 方法，异步加锁，并返回是否成功。代码如下： ``` // RedissonLock.java @Override public RFuture<Boolean> tryLockAsync(long waitTime, TimeUnit unit) { return tryLockAsync(waitTime, -1, unit); } @Override public RFuture<Boolean> tryLockAsync(long waitTime, long leaseTime, TimeUnit unit) { // 获得线程编号 long currentThreadId = Thread.currentThread().getId(); // 执行锁 return tryLockAsync(waitTime, leaseTime, unit, currentThreadId); } ``` - 最终都调用 `#tryLockAsync(long waitTime, long leaseTime, TimeUnit unit, long currentThreadId)` 方法，真正实现异步加锁的逻辑。 `#tryLockAsync(long waitTime, long leaseTime, TimeUnit unit, long currentThreadId)` 方法，代码如下： ``` // RedissonLock.java @Override public RFuture<Boolean> tryLockAsync(long waitTime, long leaseTime, TimeUnit unit, long currentThreadId) { // 创建 RPromise 对象，用于通知结果 RPromise<Boolean> result = new RedissonPromise<Boolean>(); // 表示剩余的等待获得锁的时间 AtomicLong time = new AtomicLong(unit.toMillis(waitTime)); // 记录当前时间 long currentTime = System.currentTimeMillis(); // 执行异步获得锁 RFuture<Long> ttlFuture = tryAcquireAsync(leaseTime, unit, currentThreadId); ttlFuture.onComplete((ttl, e) -> { // 如果发生异常，则通过 result 通知异常 if (e != null) { result.tryFailure(e); return; } // lock acquired // 如果获得到锁，则通过 result 通知获得锁成功 if (ttl == null) { if (!result.trySuccess(true)) { // 如果处理 result 通知对结果返回 false ，意味着需要异常释放锁 unlockAsync(currentThreadId); } return; } // 减掉已经等待的时间 long el = System.currentTimeMillis() - currentTime; time.addAndGet(-el); // 如果无剩余等待的时间，则通过 result 通知获得锁失败 if (time.get() <= 0) { trySuccessFalse(currentThreadId, result); return; } // 记录新的当前时间 long current = System.currentTimeMillis(); // 记录下面的 future 的指向 AtomicReference<Timeout> futureRef = new AtomicReference<Timeout>(); // 创建 SUBSCRIBE 订阅的 Future RFuture<RedissonLockEntry> subscribeFuture = subscribe(currentThreadId); subscribeFuture.onComplete((r, ex) -> { // 如果发生异常，则通过 result 通知异常 if (ex != null) { result.tryFailure(ex); return; } // 如果创建定时任务 Future scheduledFuture，则进行取消 if (futureRef.get() != null) { futureRef.get().cancel(); } // 减掉已经等待的时间 long elapsed = System.currentTimeMillis() - current; time.addAndGet(-elapsed); // 再次执行异步获得锁 tryLockAsync(time, leaseTime, unit, subscribeFuture, result, currentThreadId); }); // 如果创建 SUBSCRIBE 订阅的 Future 未完成，创建定时任务 Future scheduledFuture 。 if (!subscribeFuture.isDone()) { Timeout scheduledFuture = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { @Override public void run(Timeout timeout) throws Exception { // 如果创建 SUBSCRIBE 订阅的 Future 未完成 if (!subscribeFuture.isDone()) { // 进行取消 subscribeFuture subscribeFuture.cancel(false); // 通过 result 通知获得锁失败 trySuccessFalse(currentThreadId, result); } } }, time.get(), TimeUnit.MILLISECONDS); // 延迟 time 秒后执行 // 记录 futureRef 执行 scheduledFuture futureRef.set(scheduledFuture); } }); return result; } ``` - 整体逻辑是，获得分布锁。如果获取失败，则发起 Redis Pub/Sub 订阅，等待释放锁的消息，从而再次发起获得分布式锁。 - 第 11 行：调用 `#tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId)` 方法，执行异步获得锁。详细解析，胖友先跳到 [「5.1 tryAcquireAsync」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 中。 - 继续开始我们“漫长”的回调之旅。其实也比较容易懂，走起~ - 第 13 至 17 行：如果发生异常，则通过 `result` 通知异常。 - 第 19 至 26 行：如果 `ttl` 为空，说明获得到锁了，则通过 `result` 通知获得锁成功。这里，在第 23 至 24 行有个小细节，胖友自己看下注释。 - 第 41 行：声明 `futureRef` 变量，用于设置第 65 至 81 行创建的定时任务。 - 第 65 至 82 行：如果创建 SUBSCRIBE 订阅的 Future `subscribeFuture` 未完成，创建定时任务 Future `scheduledFuture` 。因为 `subscribeFuture` 是异步的，而存在一个情况，可能 `subscribeFuture` 未完成时，等待获得锁已经超时，所以通过 `scheduledFuture` 来实现超时通知。 - 第 80 行：记录 `futureRef` 为 `scheduledFuture` 。 - 第 71 行：兜底判断 `subscribeFuture` 未完成。 - 第 73 行：进行取消 `subscribeFuture` 。 - 第 75 行：调用 `#trySuccessFalse(long currentThreadId, RPromise<Boolean> result)` 方法，通知获得锁失败。代码如下： ``` // RedissonLock.java protected RFuture<Void> acquireFailedAsync(long threadId) { return RedissonPromise.newSucceededFuture(null); } private void trySuccessFalse(long currentThreadId, RPromise<Boolean> result) { acquireFailedAsync(currentThreadId).onComplete((res, e) -> { if (e == null) { // 通知获得锁失败 result.trySuccess(false); } else { // 通知异常 result.tryFailure(e); } }); } ``` - x - 第 43 至 63 行：创建 SUBSCRIBE 订阅的 Future `subscribeFuture` 。通过订阅释放锁的消息，从而实现等待锁释放的客户端，快速抢占加锁。 - 第 46 至 50 行：如果发生异常，则通过 `result` 通知异常。 - 第 52 至 55 行：如果创建定时任务 Future `scheduledFuture`，则进行取消。 - 第 57 至 59 行：减掉已经等待的时间。 - 第 62 行：调用 `#tryLockAsync(AtomicLong time, long leaseTime, TimeUnit unit, RFuture<RedissonLockEntry> subscribeFuture, RPromise<Boolean> result, long currentThreadId` 方法，再次执行异步获得锁。详细解析，见 [「5.2 更强的 tryLockAsync」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 小节。 > 艿艿：特喵的，又是一个 tryLockAsync 重载的方法，我已经瞎取标题了。深呼吸，继续！ 感叹，想要写好全异步的代码，实际是非常困难的，所以艿艿的感受，Spring Webflux 反应式框架，想要推广在编写业务逻辑，基本可能性是为零。当然，Webflux 乃至反应式编程，更加适合推广在基础组件中。 ### 5.1 tryAcquireAsync > 艿艿：看完这个方法，就跳回去哈。MMP 整个调用链，真长，大几百行代码。 `#tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId)` 方法，执行异步获得锁。代码如下： ``` // RedissonLock.java private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) { // <1> 情况一，如果锁有时长，则直接获得分布式锁 if (leaseTime != -1) { return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); } // <2> 情况二，如果锁无时长，则先获得 Lock WatchDog 的锁超时时长 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; } ``` - 一共分成两种情况，是否锁有时长。 - `<1>` 处，`leaseTime != -1` ，意味着锁设置了时长，则调用 [「3.1 `#tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command)`」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法，直接获得分布式锁。 - `<2>` 处，锁未设置了时长，所以先调用 [「3.1 `#tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command)`」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法，获得 **Lock WatchDog 的锁超时时长**的分布式锁，然后在回调中，再调用 `#scheduleExpirationRenewal(long threadId)` 方法，创建定时任务，定时调用 [「3.4 renewExpirationAsync」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) **续锁**。详细解析，见 TODO 。 ### 5.2 更强的 tryLockAsync `#tryLockAsync(AtomicLong time, long leaseTime, TimeUnit unit, RFuture<RedissonLockEntry> subscribeFuture, RPromise<Boolean> result, long currentThreadId)` 方法，**更强**的异步加锁。主要强在 2 点： - 1、增加监听锁释放的消息的监听器，从而实现等待锁的客户端快速抢占锁的逻辑。 - 2、增加锁超时自动释放，没有锁释放消息的处理。 整体代码如下： ``` // RedissonLock.java private void tryLockAsync(AtomicLong time, long leaseTime, TimeUnit unit, RFuture<RedissonLockEntry> subscribeFuture, RPromise<Boolean> result, long currentThreadId) { // 如果 result 已经完成，则直接返回，并取消订阅 if (result.isDone()) { unsubscribe(subscribeFuture, currentThreadId); return; } // 如果剩余时间 time 小于 0 ，说明等待超时，则取消订阅，并通过 result 通知失败 if (time.get() <= 0) { unsubscribe(subscribeFuture, currentThreadId); trySuccessFalse(currentThreadId, result); return; } // 记录当前时间 long curr = System.currentTimeMillis(); // 获得分布式锁 RFuture<Long> ttlFuture = tryAcquireAsync(leaseTime, unit, currentThreadId); ttlFuture.onComplete((ttl, e) -> { // 如果发生异常，则取消订阅，并通过 result 通知异常 if (e != null) { unsubscribe(subscribeFuture, currentThreadId); result.tryFailure(e); return; } // lock acquired // 如果获得到锁，则取消订阅，并通过 result 通知获得锁成功 if (ttl == null) { unsubscribe(subscribeFuture, currentThreadId); if (!result.trySuccess(true)) { unlockAsync(currentThreadId); } return; } // 减掉已经等待的时间 long el = System.currentTimeMillis() - curr; time.addAndGet(-el); // 如果无剩余等待的时间，则取消订阅，并通过 result 通知获得锁失败 if (time.get() <= 0) { unsubscribe(subscribeFuture, currentThreadId); trySuccessFalse(currentThreadId, result); return; } // waiting for message // 记录新的当前时间 long current = System.currentTimeMillis(); // 获得当前线程对应的 RedissonLockEntry 对象 RedissonLockEntry entry = getEntry(currentThreadId); // 尝试获得 entry 中的信号量，如果获得成功，说明 SUBSCRIBE 已经收到释放锁的消息，则直接立马再次去获得锁。 if (entry.getLatch().tryAcquire()) { tryLockAsync(time, leaseTime, unit, subscribeFuture, result, currentThreadId); } else { // 创建 AtomicBoolean 变量 executed ，用于标记下面创建的 listener 是否执行。 AtomicBoolean executed = new AtomicBoolean(); // 创建 AtomicReference 对象，用于指向定时任务 AtomicReference<Timeout> futureRef = new AtomicReference<Timeout>(); // 创建监听器 listener ，用于在 RedissonLockEntry 的回调，就是我们看到的 PublishSubscribe 监听到释放锁的消息，进行回调。 Runnable listener = () -> { // 标记已经执行 executed.set(true); // 如果有定时任务的 Future ，则进行取消 if (futureRef.get() != null) { futureRef.get().cancel(); } // 减掉已经等待的时间 long elapsed = System.currentTimeMillis() - current; time.addAndGet(-elapsed); // 再次获得分布式锁 tryLockAsync(time, leaseTime, unit, subscribeFuture, result, currentThreadId); }; // 添加 listener 到 RedissonLockEntry 中 entry.addListener(listener); // 下面，会创建一个定时任务。因为极端情况下，可能不存在释放锁的消息，例如说锁自动超时释放，所以需要改定时任务，在获得到锁的超时后，主动去抢下。 long t = time.get(); if (ttl >= 0 && ttl < time.get()) { // 如果剩余时间小于锁的超时时间，则使用剩余时间。 t = ttl; } // 如果 listener 未执行 if (!executed.get()) { Timeout scheduledFuture = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { @Override public void run(Timeout timeout) throws Exception { // 移除 listener 从 RedissonLockEntry 中 if (entry.removeListener(listener)) { // 减掉已经等待的时间 long elapsed = System.currentTimeMillis() - current; time.addAndGet(-elapsed); // 再次获得分布式锁 tryLockAsync(time, leaseTime, unit, subscribeFuture, result, currentThreadId); } } }, t, TimeUnit.MILLISECONDS); // 记录 futureRef 执行 scheduledFuture futureRef.set(scheduledFuture); } } }); } ``` - 第 2 至 46 行：和 [「5. tryLockAsync」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 基本一致，就不重复哔哔了。 - 第 52 行：调用 `#getEntry(long threadId)` 方法，获得当前线程对应的 RedissonLockEntry 对象。此处有点“失忆”的胖友，看看 [「4. LockPubSub」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 的结尾。 - 第 53 至 55 行：尝试获得 `entry` 中的信号量，如果获得成功，说明 SUBSCRIBE 已经收到释放锁的消息，则调用 [「5.2 `##tryLockAsync(AtomicLong time, long leaseTime, TimeUnit unit, RFuture subscribeFuture, RPromise result, long currentThreadId)`」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法，直接立马再次去获得锁。 - 第 58 行：创建 AtomicBoolean 变量 `executed` ，用于标记下面创建的 `listener` 是否执行。 - 第 60 行：声明futureRef变量，用于设置第 87 至 104 行创建的定时任务。因为极端情况下，可能不存在释放锁的消息，例如说锁自动超时释放，所以需要改定时任务，在获得到锁的超时后，主动去抢下。 - 第 82 至 85 行：计算定时任务的延迟时间时间。如果剩余时间小于锁的超时时间，则使用剩余时间。 - 第 87 行：通过 `executed` 变量，判断 `listener` 未执行。 - 第 103 行： 记录 `futureRef` 为 `scheduledFuture` 。 - 第 92 行：移除 `listener` 从 RedissonLockEntry 中。避免，可能存在的并发执行。 - 第 98 行：调用 [「5.2 `##tryLockAsync(AtomicLong time, long leaseTime, TimeUnit unit, RFuture subscribeFuture, RPromise result, long currentThreadId)`」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法，再次去获得锁。 - 这个定时任务，真的处理的是细节中的细节。之前思考获得分布式失败客户端的等待通知，只考虑了 Redis Pub/Sub 机制来实现，没有想到如果没有 PUBLISH 消息的场景。这块的逻辑，算是看 RedissonLock 最大的收获吧。 - 第 62 至 79 行：创建监听器listener，用于在 RedissonLockEntry 的回调，就是我们看到的 PublishSubscribe 监听到释放锁的消息，进行回调。 - 第 79 行：添加 `listener` 到 RedissonLockEntry 中。😈 如果胖友又“失忆”了，调回到 [「4. LockPubSub」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 再瞅瞅。 - 第 65 行：通过 `executed` 标记已经执行。 - 第 66 至 69 行：如果有定时任务的 Future ，则进行取消。 - 第 71 至 74 行：减掉已经等待的时间。 - 第 76 行：调用 [「5.2 `##tryLockAsync(AtomicLong time, long leaseTime, TimeUnit unit, RFuture subscribeFuture, RPromise result, long currentThreadId)`」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法，再次去获得锁。 ------ 至此，RedissonLock 加锁的逻辑我们已经全部看完。如果觉得略感迷糊的胖友，可以多多调试下。因为艿艿有点偷懒，未画一些图来辅助胖友理解，所以胖友可以自己画一画，嘿嘿。 ### 5.3 遗漏的 tryLockAsync 还有两个重载的 `#tryLockAsync(...)` 方法，它们是未设置锁定时长的两个。代码如下： ``` // RedissonLock.java @Override public RFuture<Boolean> tryLockAsync() { return tryLockAsync(Thread.currentThread().getId()); } @Override public RFuture<Boolean> tryLockAsync(long threadId) { return tryAcquireOnceAsync(-1, null, threadId); } ``` - 最终都调用 `#tryAcquireOnceAsync(long leaseTime, TimeUnit unit, long threadId)` 方法，真正实现异步加锁的逻辑。 `#tryAcquireOnceAsync(long leaseTime, TimeUnit unit, long threadId)` 方法，真正实现异步加锁的逻辑。代码如下： ``` // RedissonLock.java private RFuture<Boolean> tryAcquireOnceAsync(long leaseTime, TimeUnit unit, long threadId) { // 情况一，如果锁有时长，则直接获得分布式锁 if (leaseTime != -1) { return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN); } // 情况二，如果锁无时长，则先获得 Lock WatchDog 的锁超时时长 RFuture<Boolean> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN); ttlRemainingFuture.onComplete((ttlRemaining, e) -> { // 如果发生异常，则直接返回 if (e != null) { return; } // lock acquired // 如果获得到锁，则创建定时任务，定时续锁 if (ttlRemaining) { scheduleExpirationRenewal(threadId); } }); return ttlRemainingFuture; } ``` - 看到这个方法，是不是发现很熟悉，和 [「5.1 tryAcquireAsync」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 基本一模一样。差别在于它的返回的结果是 `RFuture<Boolean>` 。 - 有一点要特别注意，因为本小节我们看到的两个 `#tryLockAsync(...)` 方法，是尝试去加锁。如果加锁失败，则返回 `false` 即可，所以不会像我们在 [「5.1 tryLockAsync」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法，无限重试直到等待超时（超过 `waitTime`）。 ## 6. tryLock > 艿艿：本小节和 [「5. tryLockAsync」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 相对，为**同步**加锁。扶你起来，胖友还可以继续怼源码。 `#tryLock(long waitTime, long leaseTime, TimeUnit unit)` 方法，同步加锁，并返回是否成功。。代码如下： ``` // RedissonLock.java @Override public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException { return tryLock(waitTime, -1, unit); } @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(); // 同步获加锁 Long ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired // 加锁成功，直接返回 true 加锁成功 if (ttl == null) { return true; } // 减掉已经等待的时间 time -= System.currentTimeMillis() - current; // 如果无剩余等待的时间，则返回 false 加锁失败 if (time <= 0) { acquireFailed(threadId); return false; } // 记录新的当前时间 current = System.currentTimeMillis(); // 创建 SUBSCRIBE 订阅的 Future RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId); // 阻塞等待订阅发起成功 if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) { // 进入到此处，说明阻塞等待发起订阅超时 // 取消 SUBSCRIBE 订阅 if (!subscribeFuture.cancel(false)) { // 进入到此处，说明取消发起订阅失败，则通过设置回调，在启发订阅完成后，回调取消 SUBSCRIBE 订阅 subscribeFuture.onComplete((res, e) -> { if (e == null) { unsubscribe(subscribeFuture, threadId); } }); } // 等待超时，则返回 false 加锁失败 acquireFailed(threadId); return false; } try { // 减掉已经等待的时间 time -= System.currentTimeMillis() - current; // 如果无剩余等待的时间，则返回 false 加锁失败 if (time <= 0) { acquireFailed(threadId); return false; } while (true) { // 记录新的当前时间 long currentTime = System.currentTimeMillis(); // 同步获加锁 ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired // 加锁成功，直接返回 true 加锁成功 if (ttl == null) { return true; } // 减掉已经等待的时间 time -= System.currentTimeMillis() - currentTime; // 如果无剩余等待的时间，则返回 false 加锁失败 if (time <= 0) { acquireFailed(threadId); return false; } // waiting for message // 记录新的当前时间 currentTime = System.currentTimeMillis(); // 通过 RedissonLockEntry 的信号量，阻塞等待锁的释放消息，或者 ttl/time 超时（例如说，锁的自动超时释放） if (ttl >= 0 && ttl < time) { getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } else { getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS); } // 减掉已经等待的时间 time -= System.currentTimeMillis() - currentTime; // 如果无剩余等待的时间，则返回 false 加锁失败 if (time <= 0) { acquireFailed(threadId); return false; } } } finally { // 小细节，需要最终取消 SUBSCRIBE 订阅 unsubscribe(subscribeFuture, threadId); } // return get(tryLockAsync(waitTime, leaseTime, unit)); } ``` - 第 7 行：调用 `#tryAcquire(long leaseTime, TimeUnit unit, long threadId)` 方法，同步加锁。代码如下： ``` // RedissonLock.java private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) { return get(tryAcquireAsync(leaseTime, unit, threadId)); } ``` - 该方法内部，调用的就是 [「5.1 `#tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId)`」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法。 - 第 8 至 12 行：加锁成功，直接返回 `true` 加锁成功。 - 第 15 行：减掉已经等待的时间。 - 第 17 至 20 行：如果无剩余等待的时间，则返回 `false` 加锁失败。 - 第 25 行：调用 `#subscribe(long threadId)` 方法，创建 SUBSCRIBE 订阅的 Future `subscribeFuture` 。 - 【重要差异点】第 27 至 41 行：调用 `#await(subscribeFuture, time, TimeUnit.MILLISECONDS)` 方法，阻塞等待订阅发起成功，因为 `subscribeFuture` 是异步的，需要这一步转同步。如果发生超时，则就会进入第 28 至 37 行的取消逻辑，并在第 38 至 40 行返回 `false` 加锁失败。 - 第 52 至 89 行：反复重试，直到成功加锁返回true，或者超时返回false。 - 第 54 至 73 行：重试一波第 6 至 20 行的逻辑。 - 【重要差异点】第 75 至 80 行：通过 RedissonLockEntry 的信号量，阻塞等待锁的释放消息，或者ttl/time超时（例如说，锁的自动超时释放）。 - 相比 [「5.2 `#tryLockAsync(AtomicLong time, long leaseTime, TimeUnit unit, RFuture subscribeFuture, RPromise result, long currentThreadId)`」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法，它把信号量的等待和定时任务的等待融合在一起了。 - 等待完成后，如果无剩余时间，在第 82 至 88 行的逻辑中，返回 `false` 加锁失败。 - 等待完成后，如果有剩余时间，在第 56 行：获得重新同步获得锁。 - 第 92 行：调用 `#unsubscribe(RFuture<RedissonLockEntry> future, long threadId)` 方法，小细节，需要最终取消 SUBSCRIBE 订阅。 ## 7. lockAsync > 艿艿：本小节和 [「5. tryLockAsync」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 相似，为**异步**加锁。继续扶你起来，胖友还可以继续怼源码。 `#lockAsync(long leaseTime, TimeUnit unit, long currentThreadId)` 方法，异步加锁，无需返回是否成功。代码如下： ``` // RedissonLock.java @Override public RFuture<Void> lockAsync() { return lockAsync(-1, null); } @Override public RFuture<Void> lockAsync(long leaseTime, TimeUnit unit) { // 获得线程编号 long currentThreadId = Thread.currentThread().getId(); // 异步锁 return lockAsync(leaseTime, unit, currentThreadId); } @Override public RFuture<Void> lockAsync(long currentThreadId) { return lockAsync(-1, null, currentThreadId); } @Override public RFuture<Void> lockAsync(long leaseTime, TimeUnit unit, long currentThreadId) { // 创建 RPromise 对象，用于异步回调 RPromise<Void> result = new RedissonPromise<Void>(); // 异步加锁 RFuture<Long> ttlFuture = tryAcquireAsync(leaseTime, unit, currentThreadId); ttlFuture.onComplete((ttl, e) -> { // 如果发生异常，则通过 result 通知异常 if (e != null) { result.tryFailure(e); return; } // lock acquired // 如果获得到锁，则通过 result 通知获得锁成功 if (ttl == null) { if (!result.trySuccess(null)) { // 如果处理 result 通知对结果返回 false ，意味着需要异常释放锁 unlockAsync(currentThreadId); } return; } // 创建 SUBSCRIBE 订阅的 Future RFuture<RedissonLockEntry> subscribeFuture = subscribe(currentThreadId); subscribeFuture.onComplete((res, ex) -> { // 如果发生异常，则通过 result 通知异常 if (ex != null) { result.tryFailure(ex); return; } // 异步加锁 lockAsync(leaseTime, unit, subscribeFuture, result, currentThreadId); }); }); return result; } ``` - 第 6 行：调用 [「5.1 `#tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId)`」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法，执行异步获得锁。 - 第 7 至 35 行：又是熟悉的配方，在回调中，处理响应的加锁结果。差异就在第 34 行，见 [「7.1 更强的 lockAsync」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 的详细解析。 ## 7.1 更强的 lockAsync 实际上，`#lockAsync(long leaseTime, TimeUnit unit, RFuture<RedissonLockEntry> subscribeFuture, RPromise<Void> result, long currentThreadId)` 方法，和 [「5.2 更强的 tryLockAsync」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 是**基本一致**的。那么为什么不直接重用呢？注意，这个方法**不需要考虑等待超时**，有一种“劳资有钱，必须拿到锁”。 代码如下： ``` // RedissonLock.java private void lockAsync(long leaseTime, TimeUnit unit, RFuture<RedissonLockEntry> subscribeFuture, RPromise<Void> result, long currentThreadId) { // 获得分布式锁 RFuture<Long> ttlFuture = tryAcquireAsync(leaseTime, unit, currentThreadId); ttlFuture.onComplete((ttl, e) -> { // 如果发生异常，则取消订阅，并通过 result 通知异常 if (e != null) { unsubscribe(subscribeFuture, currentThreadId); result.tryFailure(e); return; } // lock acquired // 如果获得到锁，则取消订阅，并通过 result 通知获得锁成功 if (ttl == null) { unsubscribe(subscribeFuture, currentThreadId); if (!result.trySuccess(null)) { unlockAsync(currentThreadId); } return; } // 获得当前线程对应的 RedissonLockEntry 对象 RedissonLockEntry entry = getEntry(currentThreadId); // 尝试获得 entry 中的信号量，如果获得成功，说明 SUBSCRIBE 已经收到释放锁的消息，则直接立马再次去获得锁。 if (entry.getLatch().tryAcquire()) { lockAsync(leaseTime, unit, subscribeFuture, result, currentThreadId); } else { // waiting for message // 创建 AtomicReference 对象，用于指向定时任务 AtomicReference<Timeout> futureRef = new AtomicReference<Timeout>(); // 创建监听器 listener ，用于在 RedissonLockEntry 的回调，就是我们看到的 PublishSubscribe 监听到释放锁的消息，进行回调。 Runnable listener = () -> { // 如果有定时任务的 Future ，则进行取消 if (futureRef.get() != null) { futureRef.get().cancel(); } // 再次获得分布式锁 lockAsync(leaseTime, unit, subscribeFuture, result, currentThreadId); }; // 添加 listener 到 RedissonLockEntry 中 entry.addListener(listener); // 下面，会创建一个定时任务。因为极端情况下，可能不存在释放锁的消息，例如说锁自动超时释放，所以需要改定时任务，在获得到锁的超时后，主动去抢下。 if (ttl >= 0) { Timeout scheduledFuture = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { @Override public void run(Timeout timeout) throws Exception { // 移除 listener 从 RedissonLockEntry 中 if (entry.removeListener(listener)) { // 再次获得分布式锁 lockAsync(leaseTime, unit, subscribeFuture, result, currentThreadId); } } }, ttl, TimeUnit.MILLISECONDS); // 记录 futureRef 执行 scheduledFuture futureRef.set(scheduledFuture); } } }); } ``` - 更加熟悉的配方，全程无需处理等待锁超时的逻辑。胖友自己瞅瞅，哈哈哈。 ## 8. lock > 艿艿：本小节和 [「6. tryLoc」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 相对，为**同步**加锁。再次扶你起来，胖友还可以继续怼源码。 `#tryLock(long waitTime, long leaseTime, TimeUnit unit)` 方法，同步加锁，无需返回是否成功。代码如下： ``` // RedissonLock.java @Override public void lock() { try { lock(-1, null, false); } catch (InterruptedException e) { throw new IllegalStateException(); } } @Override public void lock(long leaseTime, TimeUnit unit) { try { lock(leaseTime, unit, false); } catch (InterruptedException e) { throw new IllegalStateException(); } } @Override public void lockInterruptibly() throws InterruptedException { lock(-1, null, true); } @Override public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException { lock(leaseTime, unit, true); } private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException { long threadId = Thread.currentThread().getId(); // 同步获加锁 Long ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired // 加锁成功，直接返回 if (ttl == null) { return; } // 创建 SUBSCRIBE 订阅的 Future RFuture<RedissonLockEntry> future = subscribe(threadId); // 阻塞等待订阅发起成功 commandExecutor.syncSubscription(future); try { while (true) { // 同步获加锁 ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired // 加锁成功，直接返回 if (ttl == null) { break; } // waiting for message // 通过 RedissonLockEntry 的信号量，阻塞等待锁的释放消息，或者 ttl/time 超时（例如说，锁的自动超时释放） if (ttl >= 0) { try { getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { // 如果允许打断，则抛出 e if (interruptibly) { throw e; } // 如果不允许打断，则继续 getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } } else { if (interruptibly) { getEntry(threadId).getLatch().acquire(); } else { getEntry(threadId).getLatch().acquireUninterruptibly(); } } } } finally { // 小细节，需要最终取消 SUBSCRIBE 订阅 unsubscribe(future, threadId); } // get(lockAsync(leaseTime, unit)); } private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) { return get(tryAcquireAsync(leaseTime, unit, threadId)); } ``` - 太过熟悉，就不哔哔了。 ------ 至此，加锁的几种组合排列，我们就已经看完了。😈 是不是有一种加锁的 Lua 脚本蛮简单的，调用 Lua 脚本实现阻塞等待的逻辑，细节还是蛮多的。如果让艿艿自己来实现这块的逻辑，估计会有一些细节处理不到位。嘿嘿。 ## 9. unlockAsync `#unlockAsync(long threadId)` 方法，异步解锁。代码如下： ``` // RedissonLock.java @Override public RFuture<Void> unlockAsync() { // 获得线程编号 long threadId = Thread.currentThread().getId(); // 执行解锁 return unlockAsync(threadId); } @Override public RFuture<Void> unlockAsync(long threadId) { // 创建 RPromise 对象，用于异步回调 RPromise<Void> result = new RedissonPromise<Void>(); // 解锁逻辑 RFuture<Boolean> future = unlockInnerAsync(threadId); future.onComplete((opStatus, e) -> { // 如果发生异常，并通过 result 通知异常 if (e != null) { cancelExpirationRenewal(threadId); result.tryFailure(e); return; } // 解锁的线程不对，则创建 IllegalMonitorStateException 异常，并通过 result 通知异常 if (opStatus == null) { IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node " + id + " thread- " + threadId); result.tryFailure(cause); return; } // 取消定时过期 cancelExpirationRenewal(threadId); // 通知 result 解锁成功 result.trySuccess(null); }); return result; } ``` - 第 7 行：调用 [3.2 `#unlockAsync(long threadId)`](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法，执行解锁逻辑。 - 第 10 至 15 行：如果发生异常，并通过 `result` 通知异常。 - 第 17 至 23 行：解锁的线程不对，则创建 IllegalMonitorStateException 异常，并通过 `result` 通知异常。😈 这里，仔细回忆下解锁 Lua 脚本的返回值。嘿嘿。 - 第 26 行：调用 `#cancelExpirationRenewal(long threadId)` 方法，取消定期过期。TODO - 第 29 行：通知 `result` 解锁成功。 ## 10. unlock `#unlock()` 方法，同步解锁。代码如下： ``` // RedissonLock.java @Override public void unlock() { try { get(unlockAsync(Thread.currentThread().getId())); } catch (RedisException e) { if (e.getCause() instanceof IllegalMonitorStateException) { throw (IllegalMonitorStateException) e.getCause(); } else { throw e; } } } ``` - 简单，基于 [「9. `#unlockAsync(long threadId)`」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法实现。 ## 11. forceUnlock `#forceUnlock()` 方法，强制解锁。代码如下： ``` // RedissonLock.java @Override public boolean forceUnlock() { // 同步 return get(forceUnlockAsync()); } @Override public RFuture<Boolean> deleteAsync() { // 异步 return forceUnlockAsync(); } ``` - 无论是同步还是异步的强制解锁，都是基于 [「3.3 `#forceUnlockAsync()`」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法实现。 ## 12. ExpirationEntry 本小节，我们来看看在 [「3.4 renewExpirationAsync」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 中，提到的**续锁**的功能。 首先，我们来看看 ExpirationEntry 类。它是 RedissonLock 的内部类，记录**续租**任务的信息。代码如下： ``` // RedissonLock.java public static class ExpirationEntry { /** * 线程与计数器的映射 * * KEY：线程编号 * VALUE：计数 */ private final Map<Long, Integer> threadIds = new LinkedHashMap<>(); /** * 定时任务 */ private volatile Timeout timeout; public ExpirationEntry() { super(); } /** * 增加线程的计数 * * @param threadId 线程编号 */ public void addThreadId(long threadId) { Integer counter = threadIds.get(threadId); if (counter == null) { counter = 1; } else { counter++; } threadIds.put(threadId, counter); } public boolean hasNoThreads() { return threadIds.isEmpty(); } public Long getFirstThreadId() { if (threadIds.isEmpty()) { return null; } return threadIds.keySet().iterator().next(); } /** * 减少线程的技术 * * @param threadId 线程编号 */ public void removeThreadId(long threadId) { Integer counter = threadIds.get(threadId); if (counter == null) { return; } counter--; if (counter == 0) { threadIds.remove(threadId); } else { threadIds.put(threadId, counter); } } public void setTimeout(Timeout timeout) { this.timeout = timeout; } public Timeout getTimeout() { return timeout; } } ``` - 可能粗略这么一看，有种然并卵的感觉，不要着急。我们下面接着看。 - 在 RedissonLock 的类中，有个EXPIRATION_RENEWAL_MAP静态属性，如下： ``` // RedissonLock.java /** * ExpirationEntry 的映射 * * key ：{@link #entryName} */ private static final ConcurrentMap<String, ExpirationEntry> EXPIRATION_RENEWAL_MAP = new ConcurrentHashMap<>(); ``` ### 12.1 scheduleExpirationRenewal `#scheduleExpirationRenewal()` 方法，发起**续锁**的定时任务。代码如下： ``` // RedissonLock.java private void scheduleExpirationRenewal(long threadId) { // 创建 ExpirationEntry 对象 ExpirationEntry entry = new ExpirationEntry(); // 添加到 EXPIRATION_RENEWAL_MAP 中 ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry); // 添加线程编号到 ExpirationEntry 中 if (oldEntry != null) { oldEntry.addThreadId(threadId); } else { entry.addThreadId(threadId); // 创建定时任务，用于续锁 renewExpiration(); } } private void renewExpiration() { // 获得 ExpirationEntry 队形，从 EXPIRATION_RENEWAL_MAP 中 ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName()); if (ee == null) { // 如果不存在，返回 return; } // 创建 Timeout 定时任务，实现定时续锁 Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { @Override public void run(Timeout timeout) throws Exception { // 获得 ExpirationEntry 对象 ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName()); if (ent == null) { // 如果不存在，返回 return; } // 获得 threadId 编号 Long threadId = ent.getFirstThreadId(); if (threadId == null) { // 如果不存在，则返回 return; } // 执行续锁 RFuture<Boolean> future = renewExpirationAsync(threadId); future.onComplete((res, e) -> { // 如果发生异常，则打印异常日志，并返回。此时，就不会在定时续租了 if (e != null) { log.error("Can't update lock " + getName() + " expiration", e); return; } // 续锁成功，则重新发起定时任务 if (res) { // reschedule itself renewExpiration(); } }); } }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); // 定时，每 internalLockLeaseTime / 3 秒执行一次。 // 设置定时任务到 ExpirationEntry 中 ee.setTimeout(task); } ``` - 第 2 至 10 行：创建 ExpirationEntry 对象，并添加到 `EXPIRATION_RENEWAL_MAP` 中，之后添加线程编号到 ExpirationEntry 中。 - 第 12 行：当且仅当entryName对应的 ExpirationEntry 对象首次创建时，才会调用#renewExpiration()方法，创建定时任务，用于续锁。 - 【重要】第 23 至 56 行：创建 Timeout 定时任务，定时每 `internalLockLeaseTime / 3` 秒执行一次**续锁**。 - 第 40 行：会调用 [「3.4 `#renewExpirationAsync(long threadId)` 方法」](http://svip.iocoder.cn/Redisson/ReentrantLock/#) 方法，执行续锁。 - 第 42 至 46 行：如果发生异常，则打印异常日志，并返回。此时，就不会在定时续租了。 - 【重要】第 48 至 52 行：如果续锁成功，则调用 `#renewExpiration()` 方法，重新发起定时任务。 - 第 59 行：设置定时任务到 ExpirationEntry 中。 ### 12.2 cancelExpirationRenewal `#cancelExpirationRenewal(Long threadId)` 方法，取消定时任务。代码如下： ``` // RedissonLock.java void cancelExpirationRenewal(Long threadId) { // 获得 ExpirationEntry 对象 ExpirationEntry task = EXPIRATION_RENEWAL_MAP.get(getEntryName()); if (task == null) { // 如果不存在，返回 return; } // 从 ExpirationEntry 中，移除线程编号 if (threadId != null) { task.removeThreadId(threadId); } // 如果 ExpirationEntry 的所有线程被清空 if (threadId == null || task.hasNoThreads()) { // 取消定时任务 task.getTimeout().cancel(); // 从 EXPIRATION_RENEWAL_MAP 中移除 EXPIRATION_RENEWAL_MAP.remove(getEntryName()); } } ``` - 当且仅当 `entryName` 对应的 `EXPIRATION_RENEWAL_MAP` 的 ExpirationEntry 对象，所有线程都被移除后，会取消定时任务。 - 整体逻辑比较简单，胖友自己瞅瞅。 ## 13. 其它方法 其它方法，比较简单，胖友自己瞅瞅即可。代码如下： ``` // RedissonLock.java @Override public Condition newCondition() { // TODO implement throw new UnsupportedOperationException(); } @Override public boolean isLocked() { return isExists(); } @Override public RFuture<Boolean> isLockedAsync() { return isExistsAsync(); } @Override public RFuture<Boolean> isExistsAsync() { return commandExecutor.writeAsync(getName(), codec, RedisCommands.EXISTS, getName()); } @Override public boolean isHeldByCurrentThread() { return isHeldByThread(Thread.currentThread().getId()); } @Override public boolean isHeldByThread(long threadId) { RFuture<Boolean> future = commandExecutor.writeAsync(getName(), LongCodec.INSTANCE, RedisCommands.HEXISTS, getName(), getLockName(threadId)); return get(future); } private static final RedisCommand<Integer> HGET = new RedisCommand<Integer>("HGET", ValueType.MAP_VALUE, new IntegerReplayConvertor(0)); public RFuture<Integer> getHoldCountAsync() { return commandExecutor.writeAsync(getName(), LongCodec.INSTANCE, HGET, getName(), getLockName(Thread.currentThread().getId())); } @Override public int getHoldCount() { return get(getHoldCountAsync()); } ``` ## 彩蛋 细节比想象中的多，代码也比想象中的多，整篇博客差不多写了 1.5 天左右。 胖友看完之后，如果还有一些细节不清晰，建议可以多多调试。总的来说，如果项目中，想要使用 Redis 分布式锁，可以考虑直接使用 Redisson 提供的 Redisson 可重入锁。可能有些胖友项目中，已经使用了 Jedis 作为 Redis 的客户端，那么可以单独使用 Redisson 来做分布式锁。 之前也和一些朋友聊过，他们项目也是采用 Jedis + Redisson 的组合，妥妥的，没问题。
评论区