锁自动续期

在tryLock的三个重载中没有指定leaseTime时，如果获取到锁，默认自动释放时间为30秒，同时每10秒自动续期。

// 尝试立即获取锁的异步方法
private RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {  
    RFuture<Long> ttlRemainingFuture;  
    // leaseTime在不指定时为-1，此时使用Reddison默认的internalLockLeaseTime（30秒）。
    // tryLockInnerAsync返回值：
    // 成功获取到锁：null的Future
    // 锁被其他线程占用：锁的ttl的Future
    if (leaseTime > 0) {  
        ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);  
    } else {  
        ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,  
                TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);  
    }  
    CompletionStage<Long> s = handleNoSync(threadId, ttlRemainingFuture);  
    ttlRemainingFuture = new CompletableFutureWrapper<>(s);  
  
    CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {  
        if (ttlRemaining == null) {  // 即获取到锁
            if (leaseTime > 0) {
                internalLockLeaseTime = unit.toMillis(leaseTime);  
            } else {  // 没有指定leaseTime，启动看门狗
                scheduleExpirationRenewal(threadId);  
            }  
        }  
        return ttlRemaining;
    });  
    return new CompletableFutureWrapper<>(f);  
}
 
// 看门狗启动入口
protected void scheduleExpirationRenewal(long threadId) {  
	// 如果当前线程的Entry不存在则创建，存在则使用旧的
    ExpirationEntry entry = new ExpirationEntry();  
    // EXPIRATION_RENEWAL_MAP是static final ConcurrentMap<String, ExpirationEntry>
    ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);  
    if (oldEntry != null) {  
        oldEntry.addThreadId(threadId);  // 重入次数加1
    } else {  
        entry.addThreadId(threadId);  // 重入次数加1
        try {  
            renewExpiration();  // 启动看门狗
        } finally {  
            if (Thread.currentThread().isInterrupted()) {  
                cancelExpirationRenewal(threadId);  
            }  
        }  
    }  
}
 
private void renewExpiration() {  
	// 检查业务是否已经完成
    ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());  
    if (ee == null) {  
        return;  
    }  
    
    // 创建Netty延时任务
    Timeout task = getServiceManager().newTimeout(new TimerTask() {  
        @Override  
        public 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;  
            }  
            // 续期脚本Future
            CompletionStage<Boolean> future = renewExpirationAsync(threadId);  
            future.whenComplete((res, e) -> { // res:续期是否成功，e:异常 
                if (e != null) {  // 有异常直接结束
                    log.error("Can't update lock {} expiration", getRawName(), e);  
                    EXPIRATION_RENEWAL_MAP.remove(getEntryName());  
                    return;  
                }  
                  
                if (res) {  
                    // 续期成功，递归
                    renewExpiration();  
                } else {  
	                // 续期失败，锁已经没了
                    cancelExpirationRenewal(null);  
                }  
            });  
        }  // internalLockLeaseTime默认为30秒，因此10秒续期一次
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);  
      
    ee.setTimeout(task);  //将定时任务存到Entry中，unlock时取消
}
 
protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {  
    return evalWriteAsync(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(getRawName()),  
            internalLockLeaseTime, getLockName(threadId));  
}