锁重试

在tryLock的三个重载中指定了waitTime时，如果立即获取锁失败，Redisson会在waitTime内进行重试。

// 尝试立即获取锁的异步方法
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);  
}
 
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {  
        // 剩余可等待时间
        long time = unit.toMillis(waitTime);  
        long current = System.currentTimeMillis();  
        long threadId = Thread.currentThread().getId();  
        // 尝试立即获取锁
        Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);   
        if (ttl == null) {  
            return true;  // 获取到锁，直接返回
        }  
          
        // 减去获取锁消耗的时间 
        time -= System.currentTimeMillis() - current;  
        if (time <= 0) {  // 剩余可等待时间不足，返回false
            acquireFailed(waitTime, unit, threadId);  
            return false;  
        }  
        
        current = System.currentTimeMillis();  
        // 订阅等待锁释放
        CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);  
        try { 
	        // 在剩余可等待时间内等待 
            subscribeFuture.get(time, TimeUnit.MILLISECONDS);  
        } catch (TimeoutException e) {  
            if (!subscribeFuture.completeExceptionally(new RedisTimeoutException(  
                    "Unable to acquire subscription lock after " + time + "ms. " +  
                            "Try to increase 'subscriptionsPerConnection' and/or 'subscriptionConnectionPoolSize' parameters."))) {  
                subscribeFuture.whenComplete((res, ex) -> {  
                    if (ex == null) {  
                        unsubscribe(res, threadId);  
                    }  
                });  
            }  
            // 剩余可等待时间内锁仍未被释放，返回false
            acquireFailed(waitTime, unit, threadId);  
            return false;  
        } catch (ExecutionException e) {  
            acquireFailed(waitTime, unit, threadId);  
            return false;  
        }  
		// 剩余可等待时间内锁释放，再次尝试获取锁
        try {  
	        // 减去等待锁释放的时间
            time -= System.currentTimeMillis() - current;  
            if (time <= 0) {  // 剩余可等待时间不足，返回false
                acquireFailed(waitTime, unit, threadId);  
                return false;  
            }  
	        
            while (true) {  
                long currentTime = System.currentTimeMillis();  
		        // 尝试获取锁
                ttl = tryAcquire(waitTime, leaseTime, unit, threadId); 
                if (ttl == null) {  
                    return true;  // 获取到锁，直接返回
                }  
                
		        // 减去获取锁消耗的时间 
                time -= System.currentTimeMillis() - currentTime;  
                if (time <= 0) {  // 剩余可等待时间不足，返回false
                    acquireFailed(waitTime, unit, threadId);  
                    return false;  
                }  
  
                currentTime = System.currentTimeMillis();  
                // 等待释放锁的消息
                if (ttl >= 0 && ttl < time) {  
	                // 锁的ttl小于剩余可等待时间，最多等到ttl，要么业务完成锁提前释放，要么锁自动过期
                    commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);  
                } else {  
	                // 锁的ttl大于等于可等待时间，最多等到time，要么业务完成锁提前释放，要么剩余可等待时间耗尽
                    commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);  
                }  
                
		        // 减去等待锁释放的时间
                time -= System.currentTimeMillis() - currentTime;  
                if (time <= 0) {  // 剩余可等待时间不足，返回false
                    acquireFailed(waitTime, unit, threadId);  
                    return false;  
                }  
                // 下一个循环继续尝试获取锁
            }  
        } finally {  
	        // 取消订阅
            unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);  
        }  
//        return get(tryLockAsync(waitTime, leaseTime, unit));  
}
    
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {  
    return commandExecutor.syncedEval(getRawName(), LongCodec.INSTANCE, command,  
            "if ((redis.call('exists', KEYS[1]) == 0) " +  
                        "or (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]);",  
            Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));  
}