江苏龙网

  1. 主页 > 生活百科 > >

一文看懂Redisson分布式锁的Watchdog机制源码实现

分布式锁

一、 分布式锁简介分布式锁是一种常见的协调分布式系统的机制，在分布式环境下保证数据的一致性和可用性 。分布式锁的实现有很多种方式，其中较为常见的方式是利用redis实现分布式锁 。
在使用 Redis 实现分布式锁时，我们通常使用 SET key value [EX seconds] [NX] 命令来给某个 key 设置一个具有过期时间的值作为锁 。其中 EX 参数表示设置过期时间，当 Redis 客户端连接断开或者达到过期时间时，锁会自动失效 。但是，在一些特殊情况下，由于网络波动等原因，我们可能无法及时续期更新锁的过期时间，这会导致锁在没有被显式释放的情况下过期，从而引发并发问题 。
Redisson 的看门狗就是为了解决这个问题而设计的 。它会在获取锁之后启动一个后台任务定期地对锁进行“续期”，即更新锁的过期时间 。具体来说，每次启动续期任务时，会通过 set(key, value, XX, PX, ttl) 命令更新锁的过期时间，同时记录该任务与锁的对应关系 。在锁释放或过期时，会取消相应的续期任务，从而保证锁的有效性 。
二、 Redisson分布式锁的看门狗源码分析Redisson 的 RedissonLock 类主要通过下面五个方法实现Watchdog机制：

  • tryAcquireAsync
  • scheduleExpirationRenewal
  • renewExpiration
  • renewExpirationAsync
  • cancelExpirationRenewal
下面讲解各个方法的源码片段
1. tryAcquireAsync
【一文看懂Redisson分布式锁的Watchdog机制源码实现】private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {// 如果设置了超时时间，调用 tryLockInnerAsync 方法尝试加锁并设置过期时间 。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;}// 如果返回值为 null 表示已经拥有锁if (ttlRemaining == null) {// 开始定期检查锁是否过期，如果没有过期则续期scheduleExpirationRenewal(threadId);}});return ttlRemainingFuture;}Redisson在lock方法执行时，会调用tryAcquireAsync方法获取锁，在获取到锁时，上面示例代码种会调用scheduleExpirationRenewal(threadId)方法，开启定时检查是否过期和自动续期的定时任务，这里其实就是看门狗机制的创建点 。
2. scheduleExpirationRenewal
private void scheduleExpirationRenewal(long threadId) {ExpirationEntry entry = new ExpirationEntry();// 在 EXPIRATION_RENEWAL_MAP 中添加一个新的键值对，如果该键名已存在，则将 threadId 添加到该键名的队列中 。ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);if (oldEntry != null) {oldEntry.addThreadId(threadId);} else { // 如果该键名不存在，则添加一个新的键值对，并执行 renewExpiration() 方法续约该锁的过期时间 。entry.addThreadId(threadId);renewExpiration();}}当一个线程获取到锁时，会调用 scheduleExpirationRenewal() 方法向 EXPIRATION_RENEWAL_MAP （ConcurrentHashMap 对象）中添加一个新的键值对，如果该键名已存在，则将 threadId 添加到该键名的队列中；否则，添加一个新的键值对，并调用 renewExpiration() 方法续约该锁的过期时间 。
3. renewExpiration
private void renewExpiration() {// 从 EXPIRATION_RENEWAL_MAP 中查找当前锁实例，并获取相应的 ExpirationEntry 对象 。ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ee == null) {return;}// 创建一个定时任务（Timeout），在线程持有该锁时执行续期操作 。Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) throws Exception {ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ent == null) {return;}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) {renewExpiration();}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); // 定时任务的间隔为锁到期时间的三分之一// 将该定时任务存储在 ExpirationEntry 实体中，方便后续处理 。ee.setTimeout(task);}


推荐阅读


上一篇：Web测试和APP测试的区别是什么？

下一篇：一文看懂Java中的ThreadLocal源码和注意事项