Redisson实现分布式锁、锁续约的案例

public class Main {
    public static void main(String[] args) throws Exception {
        // 1.配置Redis-Cluster集群节点的ip和port 
        Config config = new Config();
        config.useClusterServers()
                .addNodeAddress("redis://127.0.0.1:7001")
                .addNodeAddress("redis://127.0.0.1:7002")
                .addNodeAddress("redis://127.0.0.1:7003")
                .addNodeAddress("redis://127.0.0.1:7004");
        // 2.创建Redisson的客户端 
        RedissonClient redisson = Redisson.create(config);
        // 3.测试Redisson可重⼊锁的加锁、释放锁
        testLock(redisson);
    }

    private static void testLock(RedissonClient redisson) throws InterruptedException {
        // 1.获取key为"anyLock"的锁对象
        final RLock lock = redisson.getLock("test_lock");
        boolean locked = true;
        try {
            //2.1：加锁 
            lock.lock();
            // 2.2：加锁，并设置尝试获取锁超时时间30s、锁超时⾃动释放的时间10s 
//            locked = lock.tryLock(30, 10, TimeUnit.SECONDS);
            if (locked)
                System.out.println("加锁成功！" + new Date());
            
            Thread.sleep(20 * 1000);
            System.out.println("锁逻辑执行完毕！" + new Date());

        } finally {
            // 3.释放锁 
            lock.unlock();
        }
    }
}

RLock lock = redissonClient.getLock("test_lock");

@Override
public int calcSlot(String key) {
    if (key == null) {
        return 0;
    }

    int start = key.indexOf('{');
    if (start != -1) {
        int end = key.indexOf('}');
        if (end != -1 && start + 1 < end) {
            key = key.substring(start + 1, end);
        }
    }

    int result = CRC16.crc16(key.getBytes()) % MAX_SLOT;
    log.debug("slot {} for {}", result, key);
    return result;
}

hset test_lock UUID:ThreadId 1 
pexpire test_lock 30000

test_lock : 
{
    UUID:ThreadId：1
}

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));
}

protected void cancelExpirationRenewal(Long threadId) {
    ExpirationEntry task = EXPIRATION_RENEWAL_MAP.get(getEntryName());
    if (task == null) {
        return;
    }
    
    if (threadId != null) {
        task.removeThreadId(threadId);
    }

    if (threadId == null || task.hasNoThreads()) {
        Timeout timeout = task.getTimeout();
        if (timeout != null) {
            timeout.cancel();
        }
        EXPIRATION_RENEWAL_MAP.remove(getEntryName());
    }
}

hexists test_lock UUID:ThreadId
hincrby test_lock UUID:ThreadId 1
pexpire test_lock 30000

test_lock : 
{
    UUID:ThreadId：1
}

test_lock : 
{
    UUID:ThreadId：2
}

pttl test_lock

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
    long threadId = Thread.currentThread().getId();
    Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
    // lock acquired
    if (ttl == null) {
        return;
    }

    CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
    pubSub.timeout(future);
    RedissonLockEntry entry;
    if (interruptibly) {
        entry = commandExecutor.getInterrupted(future);
    } else {
        entry = commandExecutor.get(future);
    }

    try {
        while (true) {
            // lock() 或 lockInterruptibly()为入口走到这里时。leaseTime为-1，表示会开始开门狗；如果leaseTime大于0，则不会开启开门狗；
            ttl = tryAcquire(-1, leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                break;
            }

            // waiting for message
            if (ttl >= 0) {
                try {
                    // 因为Semaphore的可用资源为0，所以这里就等价于Thread.sleep(ttl)；
                    entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    if (interruptibly) {
                        throw e;
                    }
                    entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                }
            } else {
                if (interruptibly) {
                    entry.getLatch().acquire();
                } else {
                    entry.getLatch().acquireUninterruptibly();
                }
            }
        }
    } finally {
        unsubscribe(entry, threadId);
    }
}

@Override
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);
    // lock acquired
    if (ttl == null) {
        return true;
    }

    // 获取锁剩余的等待时长
    time -= System.currentTimeMillis() - current;
    if (time <= 0) {
        // 获取锁超时，返回获取分布式锁失败
        acquireFailed(waitTime, unit, threadId);
        return false;
    }
    
    current = System.currentTimeMillis();
    CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
    try {
        // 订阅解锁channel的超时时长为 获取锁剩余的等待时长
        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);
                }
            });
        }
        acquireFailed(waitTime, unit, threadId);
        return false;
    } catch (ExecutionException e) {
        acquireFailed(waitTime, unit, threadId);
        return false;
    }

    try {
        // 收到解锁channel的消息之后，走到这里，再次判断获取锁等待时长是否超时
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
    
        // while循环中尝试去获取锁
        while (true) {
            long currentTime = System.currentTimeMillis();
            ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                return true;
            }

            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }

            // waiting for message
            currentTime = System.currentTimeMillis();
            if (ttl >= 0 && ttl < time) {
                // 如果获取锁失败后，锁存活时长 小于 剩余锁等待时长，则线程睡眠 锁存活时长
                commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
            } else {
                // 如果获取锁失败后，锁存活时间 大于等于 剩余锁等待时长，则线程睡眠 锁等待时长
                commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
            }

            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }
        }
    } finally {
        unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);
    }
}

@Override
public RFuture<Boolean> forceUnlockAsync() {
    cancelExpirationRenewal(null);
    return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            "if (redis.call('del', KEYS[1]) == 1) then "
                    + "redis.call('publish', KEYS[2], ARGV[1]); "
                    + "return 1 "
                    + "else "
                    + "return 0 "
                    + "end",
            Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE);
}