wordpress插件整站搬家,莆田建站培训,郑州网站建设信息,最全的游戏网站目录
一、Redisson分布式锁的续期
整体分析
具体步骤和逻辑分析
为什么需要递归调用#xff1f;
定时任务的生命周期#xff1f; 一、Redisson分布式锁的续期
Redisson是一个基于Redis的Java分布式锁实现。它允许多个进程或线程之间安全地共享资源。为了实现这一点
定时任务的生命周期 一、Redisson分布式锁的续期
Redisson是一个基于Redis的Java分布式锁实现。它允许多个进程或线程之间安全地共享资源。为了实现这一点Redisson使用了一种基于分布式系统的锁机制其中锁的持有者在操作过程中需要维护锁的有效性。 关于Redisson分布式锁的详细介绍可移步到我的另一篇博客Redisson分布式锁-CSDN博客 在Redisson中锁的续期是一个关键特性用于确保在锁的持有者仍在执行任务期间锁不会被意外释放。
整体分析
锁的续期机制在Redisson中是自动管理的锁的续期是基于一个定时任务的机制定期检查锁的状态并决定是否需要续期。具体实现为 private void renewExpiration() {// 1、首先会从EXPIRATION_RENEWAL_MAP中获取一个值如果为null说明锁可能已经被释放或过期因此不需要进行续期直接返回ExpirationEntry ee EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ee null) {return;}// 2、基于TimerTask实现一个定时任务设置internalLockLeaseTime / 3的时长进行一次锁续期也就是每10s进行一次续期。Timeout task commandExecutor.getConnectionManager().newTimeout(new TimerTask() {Overridepublic void run(Timeout timeout) throws Exception {// 从EXPIRATION_RENEWAL_MAP里获取一个值检查锁是否被释放ExpirationEntry ent EXPIRATION_RENEWAL_MAP.get(getEntryName());// 如果为null则说明锁也被释放了不需要续期if (ent null) {return;}// 如果不为null则获取第一个thread也就是持有锁的线程Long threadId ent.getFirstThreadId();if (threadId null) {return;}// 如果threadId 不为null说明需要续期它会异步调用renewExpirationAsync(threadId)方法来实现续期RFutureBoolean future renewExpirationAsync(threadId);// 处理结果future.onComplete((res, e) - {// 如果有异常if (e ! null) {log.error(Cant update lock getName() expiration, e);return;}// 如果续期成功则会重新调用renewExpiration()方法进行下一次续期if (res) {// reschedule itselfrenewExpiration();}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}
具体步骤和逻辑分析
ExpirationEntry ee EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ee null) {return;}
首先从 EXPIRATION_RENEWAL_MAP 中获取当前锁的 ExpirationEntry 对象。如果该对象为null说明锁可能已经被释放或过期因此不需要进行续期直接返回。
Timeout task commandExecutor.getConnectionManager().newTimeout(new TimerTask() {Overridepublic void run(Timeout timeout) throws Exception {...}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);如果当前锁的 ExpirationEntry 对象不是null就会继续往下执行创建一个定时任务。这个定时任务的代码实现了一个锁的续期机制具体步骤和逻辑分析如下
在代码中定时任务是通过 commandExecutor.getConnectionManager().newTimeout(...) 方法创建的该任务的延迟时间设置为 internalLockLeaseTime / 3 毫秒即每次续期的时间间隔。
ExpirationEntry ent EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ent null) {return;
}在定时任务的 run 方法中首先尝试从 EXPIRATION_RENEWAL_MAP 中获取与当前锁对应的 ExpirationEntry 实例。如果获取到的 ExpirationEntry 为 null则说明锁已经被释放此时无需续期直接返回。
Long threadId ent.getFirstThreadId();
if (threadId null) {return;
}如果获取到的 ExpirationEntry 不为 null说明如果锁仍然有效继续往下走接下来获取持有该锁的线程 ID。如果 threadId 为 null也说明锁可能已经被释放直接返回。
RFutureBoolean future renewExpirationAsync(threadId);如果持有锁的线程 ID 不为 null继续往下走则调用 renewExpirationAsync(threadId) 方法异步续期锁的有效期。
继续进入这个renewExpirationAsync()方法可以看到方法的主要功能是延长锁的有效期。下面是对这段代码的详细分析 protected RFutureBoolean renewExpirationAsync(long threadId) {return 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.singletonList(getName()),internalLockLeaseTime, getLockName(threadId));}
renewExpiration()函数内部的RFutureBoolean future renewExpirationAsync(threadId);又是一个关键的函数跳入renewExpirationAsync(threadId)内部一探究竟。
返回类型RFutureBoolean 表示该方法返回一个表示异步操作结果的未来对象最终会得到一个布尔值指示续期操作是否成功。参数long threadId 是持有锁的线程 ID用于标识当前续期操作是否适用于该线程。
这个renewExpirationAsync()是一个异步刷新有效期的函数它主要是用evaLWriteAsync()方法来异步执行一段Lua脚本重置当前threadId线程持有的锁的有效期。也就是说该方法负责执行给定的Lua脚本以实现分布式锁的续期。
KEYS[1]代表锁的名称即 Redis 键。ARGV[1]引用传入的第一个非键参数表示希望设置的新过期时间毫秒锁的默认租约时间为internalLockLeaseTime。ARGV[2]引用传入的第二个非键参数表示通过getLockName(threadId)根据线程ID生成特定的锁标识符确保操作的是特定线程的锁。简单说就是持有锁的线程id。getName()获取当前锁的名称用于作为Redis中的键。LongCodec.INSTANCE编码器指示如何处理数据的序列化与反序列化。RedisCommands.EVAL_BOOLEAN表示执行的命令类型这里是执行一个返回布尔值的Lua脚本。
Lua脚本中首先执行redis.call(hexists, KEYS[1], ARGV[2]) 1该命令检查锁的名称KEYS[1]下是否存在持有该锁的线程IDARGV[1]。如果存在说明该线程仍然是锁的持有者则调用pexpire命令redis.call(pexpire, KEYS[1], ARGV[1])更新锁的过期时间。如果续期成功返回1否则返回0。
因此Lua脚本中的整体逻辑是如果当前key存在说明当前锁还被该线程持有那么就重置过期时间为30s并返回true表示续期成功反之返回false。
这段代码的设计充分利用了Redis的Lua脚本特性实现了高效且原子化的锁续期逻辑减少了并发操作中的 race condition 问题同时提供了异步执行的能力提升了系统的响应性和性能。
然后我们退回到renewExpiration()方法中继续往下走
future.onComplete((res, e) - {if (e ! null) {log.error(Cant update lock getName() expiration, e);return;}if (res) {renewExpiration();}
});通过 onComplete 方法处理续期操作的结果如果e 不为 null说明有异常则记录错误日志。如果res 为 true说明续期成功则调用 renewExpiration() 方法安排下一次的续期操作。
总结一下整体流程就是在代码中定时任务是通过 commandExecutor.getConnectionManager().newTimeout(...) 方法创建的。该任务会在指定的时间internalLockLeaseTime / 3 毫秒后执行一次。每当任务执行时都会检查当前锁的状态并尝试续期。如果需要续期即锁仍然有效则会调用 renewExpiration() 方法。
为什么需要递归调用
在锁的实现中为了确保锁在持有者处理任务期间保持有效通常会设置一个有效期lease time。在有效期内如果持有锁的线程仍然在执行任务那么它需要定期续期以防止在任务完成前锁过期从而导致其他线程获取锁。
递归调用的机制在 run 方法的最后如果续期成功调用 renewExpiration() 方法。这通常意味着该方法会重新安排另一个定时任务相当于在每次续期后再次创建一个新的定时任务使得续期操作可以持续进行。这种递归调用的方式确保了只要锁仍然被持有续期操作就会不断地被调度从而保持锁的有效性。
定时任务的生命周期
每个定时任务的生命周期是短暂的完成一次 run 方法的执行后该任务就结束了。然后通过递归调用可能会创建新的定时任务从而继续续期。
1任务通过 newTimeout 被创建并且首次执行会在 internalLockLeaseTime / 3 毫秒后触发。这个时间间隔确保了任务在锁的生命周期的早期进行检查和续期。此时任务进入其生命周期准备执行。
2当定时任务第一次执行时run() 方法被调用。它主要的任务是
从 EXPIRATION_RENEWAL_MAP 获取锁的状态。如果锁被释放ent null任务直接返回不再进行续期。如果锁仍然存在并且当前线程持有锁threadId ! null则异步调用 renewExpirationAsync(threadId) 来续期锁。在续期的异步任务完成后如果续期成功res true会重新调用 renewExpiration() 进行下一次续期。
3续期条件如果任务成功续期它会在异步任务的 onComplete 回调中再次调用 renewExpiration() 方法。renewExpiration() 负责创建一个新的定时任务这意味着每次任务续期成功后系统会重新调度一个新的定时任务以确保锁的有效期能够持续。
这个 renewExpiration() 方法的调用实际上是递归调用新的定时任务续期继续进行下去。每次任务执行后都可能会创建一个新的任务直到锁被释放。
3定时任务的生命周期可能在以下情况下终止
锁被释放当 EXPIRATION_RENEWAL_MAP.get(getEntryName()) 返回 null表示锁已经被释放定时任务会停止续期不再创建新的定时任务。无持有锁的线程如果没有线程持有锁即 threadId null任务也会停止续期。异步任务失败如果续期的异步任务失败例如网络问题、数据库问题等则可能无法继续续期。不过在代码中如果发生异常它只会记录错误并不会立即停止整个续期机制但最终续期将会失败并终止。
定时任务的生命周期从它的创建开始通过定期执行检查和续期直到锁被释放或没有线程持有锁时任务才会停止。每次续期成功后新的定时任务会继续执行确保锁的有效期在持锁线程存在时不会过期。
因此虽然定时任务会被创建并执行但它的执行是基于持锁状态的只有在锁有效且持有者仍在执行任务的情况下才会持续进行续期。这个设计确保了资源的有效管理避免不必要的续期操作。
