Redisson加锁和解锁、WatchDog机制的原理

Redisson加锁和解锁、WatchDog机制的原理

一、加锁

默认加锁方法：RLock#lock()

redisson通过lua脚本来保证加锁的原子性，用客户端对应的线程的唯一标识来保证加锁的用户不被抢占，用过期时间和WatchDog机制（可选）保证不死锁。

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
    long threadId = Thread.currentThread().getId();
    // 尝试加锁，并返回锁的剩余时间
    Long ttl = tryAcquire(leaseTime, unit, threadId);
    // lock acquired
    if (ttl == null) {
        // 返回null，表示加锁成功
        return;
    }

    // 加锁失败，订阅redisson_lock__channel:{锁名称}频道来获取锁的释放消息
    // protected RFuture subscribe(long threadId) {
    //    return pubSub.subscribe(getEntryName(), getChannelName());
    // }
    RFuture future = subscribe(threadId);
    commandExecutor.syncSubscription(future);

    try {
        // 收到锁的释放消息，自旋获取锁
        while (true) {
            ttl = tryAcquire(leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                break;
            }

            // waiting for message
            if (ttl >= 0) {
                try {
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    if (interruptibly) {
                        throw e;
                    }
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                }
            } else {
                if (interruptibly) {
                    getEntry(threadId).getLatch().acquire();
                } else {
                    getEntry(threadId).getLatch().acquireUninterruptibly();
                }
            }
        }
    } finally {
        // 无论是否获取锁成功，都取消订阅
        unsubscribe(future, threadId);
    }
}

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
    return get(tryAcquireAsync(leaseTime, unit, threadId));
}

private  RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
    if (leaseTime != -1) {
    	// 如果客户端自定义过期时间，则不适用WatchDog机制
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    // WatchDog机制看下面分析
    RFuture ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
        if (e != null) {
            return;
        }

        // lock acquired
        if (ttlRemaining == null) {
            scheduleExpirationRenewal(threadId);
        }
    });
    return ttlRemainingFuture;
}

 RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {
    internalLockLeaseTime = unit.toMillis(leaseTime);

    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                                          "if (redis.call('exists', KEYS[1]) == 0) then " +
                                          "redis.call('hset', 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]);",
                                          Collections.