当前位置：首页 > news >正文

Redisson Watchdog实现原理与源码解析：分布式锁的自动续期机制

news 来源：原创 2025/4/24 8:10:23

引言

在分布式系统中，Redis分布式锁是解决资源竞争问题的常用方案。然而，当持有锁的客户端因GC、网络延迟或处理时间过长导致锁过期时，可能引发数据一致性问题。Redisson的Watchdog（看门狗）机制通过自动续期解决了这一痛点。本文将深入分析其实现原理，结合源码揭示其工作机制。

一、Watchdog的核心原理

1.1 什么是Watchdog？

Watchdog是Redisson客户端的一个后台线程，用于监控并自动续期分布式锁。当客户端获取锁时未显式指定leaseTime（锁持有时间），Watchdog会启动，定期（默认每10秒）检查锁状态并延长锁的过期时间，避免业务未完成时锁提前释放。

1.2 核心机制

锁续期条件：仅当未指定leaseTime时生效。
续期规则：默认锁超时时间为30秒，每次续期重置为30秒。
续期间隔：超时时间的1/3（即10秒）。
自动释放：客户端宕机或进程退出时，Watchdog线程终止，锁超时后自动释放，避免死锁。

二、源码分析：从加锁到续期

2.1 关键类与入口

Redisson的分布式锁实现位于RedissonLock类中。核心方法为：

// 获取锁入口
void lock(long leaseTime, TimeUnit unit);

2.2 锁获取与Watchdog启动

当leaseTime未设置（默认为-1）时，触发Watchdog初始化：

// RedissonLock.java
public void lock() {try {lock(-1, null, false);} // ...
}private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {long threadId = Thread.currentThread().getId();Long ttl = this.tryAcquire(-1L, leaseTime, unit, threadId);if (ttl != null) {CompletableFuture<RedissonLockEntry> future = this.subscribe(threadId);this.pubSub.timeout(future);RedissonLockEntry entry;if (interruptibly) {entry = (RedissonLockEntry)this.commandExecutor.getInterrupted(future);} else {entry = (RedissonLockEntry)this.commandExecutor.get(future);}try {while(true) {ttl = this.tryAcquire(-1L, leaseTime, unit, threadId);if (ttl == null) {return;}if (ttl >= 0L) {try {entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);} catch (InterruptedException var14) {InterruptedException e = var14;if (interruptibly) {throw e;}entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);}} else if (interruptibly) {entry.getLatch().acquire();} else {entry.getLatch().acquireUninterruptibly();}}} finally {this.unsubscribe(entry, threadId);}}}

2.3 续期任务调度

scheduleExpirationRenewal()启动定时任务：

// RedissonBaseLock.java
protected void scheduleExpirationRenewal(long threadId) {ExpirationEntry entry = new ExpirationEntry();ExpirationEntry oldEntry = (ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName(), entry);if (oldEntry != null) {oldEntry.addThreadId(threadId);} else {entry.addThreadId(threadId);try {this.renewExpiration();} finally {if (Thread.currentThread().isInterrupted()) {this.cancelExpirationRenewal(threadId);}}}}private void renewExpiration() {ExpirationEntry ee = (ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());if (ee != null) {Timeout task = this.commandExecutor.getConnectionManager().newTimeout(new TimerTask() {public void run(Timeout timeout) throws Exception {ExpirationEntry ent = (ExpirationEntry)RedissonBaseLock.EXPIRATION_RENEWAL_MAP.get(RedissonBaseLock.this.getEntryName());if (ent != null) {Long threadId = ent.getFirstThreadId();if (threadId != null) {CompletionStage<Boolean> future = RedissonBaseLock.this.renewExpirationAsync(threadId);future.whenComplete((res, e) -> {if (e != null) {RedissonBaseLock.log.error("Can't update lock " + RedissonBaseLock.this.getRawName() + " expiration", e);RedissonBaseLock.EXPIRATION_RENEWAL_MAP.remove(RedissonBaseLock.this.getEntryName());} else {if (res) {RedissonBaseLock.this.renewExpiration();} else {RedissonBaseLock.this.cancelExpirationRenewal((Long)null);}}});}}}}, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);ee.setTimeout(task);}}

2.4 续期操作实现

renewExpirationAsync()通过Lua脚本延长锁的过期时间：

  // RedissonBaseLock.javaprotected CompletionStage<Boolean> renewExpirationAsync(long threadId) {return this.evalWriteAsync(this.getRawName(), 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(this.getRawName()), this.internalLockLeaseTime, this.getLockName(threadId));}