Redisson 分布式锁（二）lock 续约操作以及对应的lua脚本

喜欢ヅ旅行 2022-10-31 13:53 2阅读 0赞

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
            if (leaseTime != -1) {
                return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
            }
            //这里是什么？这里的RFuture ttlRemainingFuture  封装的就是当前的key 剩余存活的时间还有多少，
            RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
            //向ttlRemainingFuture  中注册一个监听器 干什么用？监听异步上锁执行
            ttlRemainingFuture.addListener(new FutureListener<Long>() {
                @Override
                public void operationComplete(Future<Long> future) throws Exception {
                    if (!future.isSuccess()) {
                        return;
                    }
                    //获取返回值
                    Long ttlRemaining = future.getNow();
                    // lock acquired
                    //如果返回了key 剩余存活时间那么 ttlRemaining 肯定不为null
                    //说明上锁成功，但是我们肯定要进行续约对吧，所以这里看名称 scheduleExpirationRenewal 调度过期时间的刷新，其实就是续约，开启异步线程对锁进行续约，也就是我们提到的watchdog看门狗
                    if (ttlRemaining == null) {
                        scheduleExpirationRenewal(threadId);
                    }
                }
            });

这里我们看到向ttlRemainingFuture  中注册了一个监听器，那到底是什么用？其实很简单，监听嘛，肯定是当我们的lua 脚本执行完成后，触发监听，执行相应的逻辑，，

（1）如果返回一个Long 对象 不为null，那么说明加锁失败，烦货的Long 值就是我们对应key 的剩余存活时间

（2）返回Long 对象ttlRemainng 为null  说明加锁成功，开启一个异步线程，定时调度刷新key的存活时间（watchdog）

private void scheduleExpirationRenewal(final long threadId) {
            //这里其实就是一个幂等处理，思考 如果存在了当前线程的加锁调式调度，肯定就不需要再添加一次对把
            if (expirationRenewalMap.containsKey(getEntryName())) {
                return;
            }
               //开启一个异步线程 默认 30/3 = 10S 后调度任务
                //进行续约，如果线程没有持有锁了，那么就会停止调度
            Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
                @Override
                public void run(Timeout timeout) throws Exception {
                       //从名字可以看出这里是异步的更新过期时间
                    RFuture<Boolean> future = renewExpirationAsync(threadId);
                    
                    future.addListener(new FutureListener<Boolean>() {
                        @Override
                        public void operationComplete(Future<Boolean> future) throws Exception {
                            expirationRenewalMap.remove(getEntryName());
                            if (!future.isSuccess()) {
                                log.error("Can't update lock " + getName() + " expiration", future.cause());
                                return;
                            }
                            //0 false  1 true  意思就是 只有成功了才会再次调度续约
                            if (future.getNow()) {
                                // reschedule itself
                                //这里续约完成之后，会再次调用 续约方法
                                scheduleExpirationRenewal(threadId);
                            }
                        }
                    });
                }
                //过期时间/3 = 默认就是 30 / 3 = 10S
            }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
    
            if (expirationRenewalMap.putIfAbsent(getEntryName(), new ExpirationEntry(threadId, task)) != null) {
                task.cancel();
            }
        }

protected RFuture<Boolean> renewExpirationAsync(long threadId) {
            return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                       //先判断是否对应线程 是否持有锁，
                        //KEY[1] 是锁名称 ARGV[2]是唯一的标识 UUID:ThreadId
                    "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        //如果持有锁，那么就重新设置过期时间为 30S
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return 1; " +
                    "end; " +
                    "return 0;",
                Collections.<Object>singletonList(getName()), 
                internalLockLeaseTime, getLockName(threadId));
        }

if (redis.call('hexists', KEYS\[1\], ARGV\[2\]) == 1)

> 先判断是否对应线程 是否持有锁，  
> KEY\[1\] 是锁名称 ARGV\[2\]是唯一的标识 UUID:ThreadId

redis.call('pexpire', KEYS\[1\], ARGV\[1\])

> 如果持有锁，那么就重新设置过期时间为 30S

所以这里就是一个异步的线程定时 过期时间/3 默认10S 去将key 的过期时间重新设置为30S

当然如果对应持有锁的机器宕机了，那么redis 中的key 就会最多30S之后才会释放锁，30S的锁，已经很久了，这也是Redisson 一个不太友好的操作。

接下来我们继续看下如何手动释放锁