分布式锁的最佳实践之Redisson

本文接从库存超卖问题分析锁和分布式锁的应用（二）讲解Redisson在分布式锁的应用实践。

官网文档地址：https://github.com/redisson/redisson/wiki

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最简单和最便捷的方法。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。

pom依赖：

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.17.1</version>
</dependency>

【1】可重入锁（Reentrant Lock）

基于Redis的分布式可重入锁对象用于Java，并实现了Lock接口。它使用发布/订阅（pub/sub）通道来通知所有等待获取锁的Redisson实例中的其他线程。

如果获得锁的Redisson实例崩溃，那么该锁可能会永远处于已获取状态，导致死锁。为了避免这种情况，Redisson维护了一个锁看门狗（lock watchdog），只要持有锁的Redisson实例还活着，它就会延长锁的过期时间。默认情况下，锁看门狗的超时时间为30秒，并且可以通过Config.lockWatchdogTimeout设置进行更改。

在获取锁时可以定义leaseTime参数。在指定的时间间隔后，被锁定的锁将自动释放。

RLock对象根据Java的Lock规范行为。这意味着只有锁的所有者线程才能解锁，否则会抛出IllegalMonitorStateException异常。如果需要多个线程可以同时拥有锁，则应考虑使用RSemaphore对象。

看门狗、leaseTime以及只有锁的所有者线程才能解锁对本文所有锁章节都适用。

简而言之，Redisson的RLock提供了一种机制，使得在分布式环境中多个节点能够安全地共享和控制对资源的访问，而不会出现死锁或竞争条件。通过使用Redis作为协调服务，它确保了锁的一致性和可用性，即使在某个节点失败的情况下也能维持系统运行的稳定性。

RLock lock = redisson.getLock("anyLock");
// 最常见的使用方法
lock.lock();

// 加锁以后10秒钟自动解锁// 无需调用unlock方法手动解锁
lock.lock(10, TimeUnit.SECONDS);

// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);

① 配置

官网文档配置章节详细说明了单节点模式、哨兵模式、集群模式等情况下关于Redisson的配置。

单节点模式配置如下：

@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient(){
        Config config = new Config();
        // 可以用"rediss://"来启用SSL连接
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        //... 还可以设置其他参数比如用户名、密码、连接数等
        return Redisson.create(config);
    }
}

② 锁实践

修改我们的代码如下：

@Autowired
private RedissonClient redissonClient;

public void deduct() {
    // 加锁，获取锁失败重试
    RLock lock = this.redissonClient.getLock("lock");
    lock.lock();

    try {
        // 1. 查询库存信息
        String stock = redisTemplate.opsForValue().get("stock");
        // 2. 判断库存是否充足
        if (stock != null && stock.length() != 0) {
            int st = Integer.parseInt(stock);
            if (st > 0) {
                // 3.扣减库存
                redisTemplate.opsForValue().set("stock", String.valueOf(--st));
            }
        }
    } finally {
        // 释放锁
        lock.unlock();
    }
}

这里获取的是RedissonLock实例，其实现了可重入（底层是lua脚本）和定期续约（底层是时间轮定时器+lua脚本）功能。其解锁的核心代码如下所示，在解锁成功后采用了发布/订阅的机制来通知其他阻塞该锁的线程来获取锁。

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
   return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
           "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                   "return nil;" +
                   "end; " +
                   "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                   "if (counter > 0) then " +
                   "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                   "return 0; " +
                   "else " +
                   "redis.call('del', KEYS[1]); " +
                   "redis.call('publish', KEYS[2], ARGV[1]); " +
                   "return 1; " +
                   "end; " +
                   "return nil;",
           Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}

③ 看门狗如何解决死锁问题

当一个Redisson实例获得了锁并突然崩溃，其内部的锁看门狗机制会防止锁永久持有，从而避免潜在的死锁情况。看门狗的工作原理如下：

1. 初始设置锁过期时间：
   当一个Redisson实例成功获取锁时，它会在Redis中为这个锁设置一个过期时间（TTL）。这个过期时间是锁的租约时间，即leaseTime参数，加上一个额外的安全缓冲时间，通常由lockWatchdogTimeout配置项决定，默认是30秒。

2. 看门狗续租：
   成功获取锁的Redisson实例会启动一个后台线程，即看门狗，它会周期性地检查锁的持有者是否仍然活跃。如果Redisson实例正常运行，看门狗会在锁的过期时间到达之前，自动更新锁的过期时间，以保持锁的有效性。这个操作会持续进行，直到锁的持有者主动释放锁或Redisson实例本身停止运行。

3. 检测实例崩溃：
   如果持有锁的Redisson实例崩溃，它的看门狗也将随之停止工作，不再能更新锁的过期时间。因此，一旦超过锁的总过期时间（leaseTime + lockWatchdogTimeout），锁将在Redis中自动过期并被释放。

4. 其他实例尝试获取锁：
   锁一旦释放，其他正在等待的Redisson实例可以尝试获取这个锁。由于锁的过期，它们将有机会成为新的锁持有者，从而继续执行其任务，避免了死锁的发生。

通过这种方式，即使某个Redisson实例崩溃，系统也能够自动恢复并允许其他实例继续正常工作，确保分布式系统的健壮性和一致性。

【2】公平锁

基于Redis的Redisson分布式可重入公平锁也是实现了 java.util.concurrent.locks.Lock 接口的一种 RLock 对象。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

它保证了当多个Redisson客户端线程同时请求加锁时，优先分配给先发出请求的线程。所有请求线程会在一个队列中排队，当某个线程出现宕机时，Redisson会等待5秒后继续下一个线程，也就是说如果前面有5个线程都处于等待状态，那么后面的线程会等待至少25秒。

其同样可以指定过期时间，在获取锁时，可以定义leaseTime参数。在指定的时间间隔后，被锁定的锁将自动释放。

RLock lock = redisson.getFairLock("myLock");

// traditional lock method
lock.lock();

// or acquire lock and automatically unlock it after 10 seconds
lock.lock(10, TimeUnit.SECONDS);

// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
   try {
     ...
   } finally {
       lock.unlock();
   }
}

【3】红锁

基于Redis的Redisson红锁 RedissonRedLock 对象实现了Redlock介绍的加锁算法。该对象也可以用来将多个 RLock 对象关联为一个红锁，每个 RLock 对象实例可以来自于不同的Redisson实例。

RLock lock1 = redissonInstance1.getLock("lock1");
RLock lock2 = redissonInstance2.getLock("lock2");
RLock lock3 = redissonInstance3.getLock("lock3");
RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
// 同时加锁：lock1 lock2 lock3
// 红锁在大部分节点上加锁成功就算成功。
lock.lock();
...
lock.unlock();

目前已经过期，被 RLock or RFencedLock 替代。

【4】读写锁

基于Redis的分布式可重入读写锁（ReadWriteLock）对象为Java实现了ReadWriteLock接口。其中，读锁（ReadLock）和写锁（WriteLock）都实现了RLock接口。

该读写锁允许多个读锁持有者和仅一个写锁持有者存在。

如果获得锁的Redisson实例崩溃，那么该锁可能永远保持在获取状态，导致死锁。为了避免这种情况，Redisson维护了一个锁看门狗，只要持有锁的Redisson实例还活着，它就会延长锁的过期时间。默认情况下，锁看门狗的超时时间是30秒，这个值可以通过Config.lockWatchdogTimeout设置来更改。

此外，Redisson允许在获取锁时指定leaseTime参数。在指定的时间间隔后，被锁定的锁将自动释放。

RLock对象遵循Java的Lock规范。这意味着只有锁的所有者线程才能解锁，否则会抛出IllegalMonitorStateException异常。如果需要多个线程可以同时拥有锁，那么应该考虑使用RSemaphore对象。

总结来说，Redisson提供的分布式读写锁确保了读操作可以并发进行，而写操作具有排他性。通过锁看门狗和租约时间的机制，它不仅提供了锁的自动释放功能，还确保了即使在持有锁的实例发生故障时，系统也能避免死锁，保持一致性和可用性。

RReadWriteLock rwlock = redisson.getReadWriteLock("myLock");

// RedissonReadLock
RLock lock = rwlock.readLock();

// RedissonWriteLock
RLock lock = rwlock.writeLock();

// traditional lock method
lock.lock();

// or acquire lock and automatically unlock it after 10 seconds
lock.lock(10, TimeUnit.SECONDS);

// or wait for lock aquisition up to 100 seconds 
// and automatically unlock it after 10 seconds
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
   try {
     ...
   } finally {
       lock.unlock();
   }
}

比如我们有两个请求，一个读，一个写，分别对应不同的方法逻辑：

public String testRead() {
    RReadWriteLock rwLock = this.redissonClient.getReadWriteLock("rwLock");
    rwLock.readLock().lock(10, TimeUnit.SECONDS);

    System.out.println("测试读锁。。。。");
    // rwLock.readLock().unlock();

    return null;
}

public String testWrite() {
    RReadWriteLock rwLock = this.redissonClient.getReadWriteLock("rwLock");
    rwLock.writeLock().lock(10, TimeUnit.SECONDS);

    System.out.println("测试写锁。。。。");
    // rwLock.writeLock().unlock();

    return null;
}

打开两个浏览器窗口测试：

• 同时访问写：一个写完之后，等待一会儿（约10s），另一个写开始
• 同时访问读：不用等待
• 先写后读：读要等待（约10s）写完成
• 先读后写：写要等待（约10s）读完成

【5】信号量

基于Redis的Redisson的分布式信号量（Semaphore）Java对象 RSemaphore 采用了与java.util.concurrent.Semaphore 相似的接口和用法。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

JUC的Semaphore可以解决服务实例内部/方法内部的资源竞争问题，但是无法解决分布式场景下的资源竞争问题。Redisson的分布式信号量RSemaphore 可以解决分布式场景下资源竞争问题。

示例如下：

RSemaphore semaphore = redisson.getSemaphore("mySemaphore");

//尝试设置许可数量，如果设置成功返回true，如果已经被设置过，返回false。
boolean  res = trySetPermits(3);

// acquire single permit 获取一个许可
semaphore.acquire();

// or acquire 10 permits 获取10个许可
semaphore.acquire(10);

// or try to acquire permit 尝试获取一个许可
boolean res = semaphore.tryAcquire();

// or try to acquire permit or wait up to 15 seconds 尝试获取许可，15秒后自动放弃
boolean res = semaphore.tryAcquire(15, TimeUnit.SECONDS);

// or try to acquire 10 permit 尝试获取10个许可
boolean res = semaphore.tryAcquire(10);

// or try to acquire 10 permits or wait up to 15 seconds
//尝试获取10个许可，等待15秒后自动放弃
boolean res = semaphore.tryAcquire(10, 15, TimeUnit.SECONDS);
if (res) {
   try {
     ...
   } finally {
	   // 释放资源
       semaphore.release();
   }
}

【6】闭锁（CountDownLatch）

基于Redisson的Redisson分布式闭锁（CountDownLatch）Java对象RCountDownLatch采用了与java.util.concurrent.CountDownLatch相似的接口和用法。

在使用之前应通过trySetCount(count)方法进行初始化。

代码使用示例如下：

RCountDownLatch latch = redisson.getCountDownLatch("myCountDownLatch");

latch.trySetCount(1);

// await for count down -阻塞等待计数减减
latch.await();

// in other thread or JVM
RCountDownLatch latch = redisson.getCountDownLatch("myCountDownLatch");
// 计数-1 ，为0 时会唤醒阻塞线程
latch.countDown();

闭锁又称倒计时器，适用于某个线程等待一组子线程完成（或者是发生、处理中）任务的场景。