Redisson 实现分布式锁的原理分析 (3)

注意:从以上源码我们看到 leaseTime 必须是 -1 才会开启 Watch Dog 机制,也就是如果你想开启 Watch Dog 机制必须使用默认的加锁时间为 30s。如果你自己自定义时间,超过这个时间,锁就会自定释放,并不会延长。

private void scheduleExpirationRenewal(long threadId) { ExpirationEntry entry = new ExpirationEntry(); ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry); if (oldEntry != null) { oldEntry.addThreadId(threadId); } else { entry.addThreadId(threadId); renewExpiration(); } } 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 " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return 1; " + "end; " + "return 0;", Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); }

Watch Dog 机制其实就是一个后台定时任务线程,获取锁成功之后,会将持有锁的线程放入到一个 RedissonLock.EXPIRATION_RENEWAL_MAP里面,然后每隔 10 秒 (internalLockLeaseTime / 3) 检查一下,如果客户端 1 还持有锁 key(判断客户端是否还持有 key,其实就是遍历 EXPIRATION_RENEWAL_MAP 里面线程 id 然后根据线程 id 去 Redis 中查,如果存在就会延长 key 的时间),那么就会不断的延长锁 key 的生存时间。

注意:这里有一个细节问题,如果服务宕机了,Watch Dog 机制线程也就没有了,此时就不会延长 key 的过期时间,到了 30s 之后就会自动过期了,其他线程就可以获取到锁。

可重入加锁机制

Redisson 也是支持可重入锁的,比如下面这种代码:

@Override public void lock() { RLock lock = redissonSingle.getLock("myLock"); try { lock.lock(); // 执行业务 doBusiness(); lock.lock(); } catch (Exception e) { e.printStackTrace(); } finally { // 释放锁 lock.unlock(); lock.unlock(); logger.info("任务执行完毕, 释放锁!"); } }

我们再分析一下加锁那段 lua 代码:

if (redis.call('exists', KEYS[1]) == 0) then " + "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "return redis.call('pttl', KEYS[1]);"

第一个 if 判断肯定不成立,exists myLock 会显示锁 key 已经存在。第二个 if 判断会成立,因为 myLock 的 hash 数据结构中包含的那个 ID 即客户端 1 的 ID,此时就会执行可重入加锁的逻辑,使用:hincrby myLock 285475da-9152-4c83-822a-67ee2f116a79:52 1 对客户端 1 的加锁次数加 1。此时 myLock 数据结构变为下面这样:

127.0.0.1:6379> HGETALL myLock 1) "285475da-9152-4c83-822a-67ee2f116a79:52" 2) "2"

到这里,小伙伴本就都明白了 hash 结构的 key 是锁的名称,field 是客户端 ID,value 是该客户端加锁的次数。

这里有一个细节,如果加锁支持可重入锁,那么解锁呢?

释放锁机制

执行

lock.unlock()

就可以释放分布式锁。我们来看一下释放锁的流程代码:

@Override public RFuture<Void> unlockAsync(long threadId) { RPromise<Void> result = new RedissonPromise<Void>(); // 1. 异步释放锁 RFuture<Boolean> future = unlockInnerAsync(threadId); // 取消 Watch Dog 机制 future.onComplete((opStatus, e) -> { cancelExpirationRenewal(threadId); if (e != null) { result.tryFailure(e); return; } if (opStatus == null) { IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + id + " thread-id: " + threadId); result.tryFailure(cause); return; } result.trySuccess(null); }); return result; } protected RFuture<Boolean> unlockInnerAsync(long threadId) { return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, // 判断锁 key 是否存在 "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + "return nil;" + "end; " + // 将该客户端对应的锁的 hash 结构的 value 值递减为 0 后再进行删除 // 然后再向通道名为 redisson_lock__channel publish 一条 UNLOCK_MESSAGE 信息 "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " + "if (counter > 0) then " + "redis.call('pexpire', KEYS[1], ARGV[2]); " + "return 0; " + "else " + "redis.call('del', KEYS[1]); " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; "+ "end; " + "return nil;", Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId)); }

从以上代码来看,释放锁的步骤主要分三步:

删除锁(这里注意可重入锁,在上面的脚本中有详细分析)。

广播释放锁的消息,通知阻塞等待的进程(向通道名为 redisson_lock__channel publish 一条 UNLOCK_MESSAGE 信息)。

取消 Watch Dog 机制,即将 RedissonLock.EXPIRATION_RENEWAL_MAP 里面的线程 id 删除,并且 cancel 掉 Netty 的那个定时任务线程。

方案优点

Redisson 通过 Watch Dog 机制很好的解决了锁的续期问题。

和 Zookeeper 相比较,Redisson 基于 Redis 性能更高,适合对性能要求高的场景。

内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:https://www.heiqu.com/wssfsd.html