Redisson分布式锁和同步器及源码分析

雨点打透心脏的1/2处 2023-07-17 04:34 42阅读 0赞

##### 1.可重入锁（Reentrant Lock） #####

基于Redis的Redisson分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口。同时还提供了异步、反射式和RxJava2标准的接口。

RLock lock = redisson.getLock("anyLock");
    // 最常见的使用方法
    lock.lock();
    //尝试加锁，
    boolean flag = lock.tryLock(32, TimeUnit.SECONDS);

大家都知道，如果负责储存这个分布式锁的Redisson节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改[Config.lockWatchdogTimeout][]来另行指定。

另外Redisson还通过加锁的方法提供了`leaseTime`的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

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

Redisson同时还为分布式锁提供了异步执行的相关方法：

RLock lock = redisson.getLock("anyLock");
    lock.lockAsync();
    lock.lockAsync(10, TimeUnit.SECONDS);
    Future<Boolean> res = lock.tryLockAsync(100, 10, TimeUnit.SECONDS);

`RLock`对象完全符合Java的Lock规范。也就是说只有拥有锁的进程才能解锁，其他进程解锁则会抛出`IllegalMonitorStateException`错误。但是如果遇到需要其他进程也能解锁的情况，请使用[分布式信号量`Semaphore`][Semaphore] 对象.

在学习这一块的时候遇到看门狗对锁不续期的问题，那这个又是什么原因呢？看如下源码：

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) { 
            //租约时间非-1
            if (leaseTime != -1) { 
                return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
            }
            //过期时间-1，也就是默认时间是30s时执行异步任务
            RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
            //异步任务执行成功后的回调函数
            ttlRemainingFuture.onComplete((ttlRemaining, e) -> { 
            		//异常直接返回
                if (e != null) { 
                    return;
                }
    
                // lock acquired 加锁成功返回null
                if (ttlRemaining == null) { 
                		//定时调度任务给锁续期
                    scheduleExpirationRenewal(threadId);
                }
            });
            return ttlRemainingFuture;
        }

通过上面的代码我们可以看出，只要租约时间leaseTime的值不是-1就不会开启看门狗的定时任务；直到了原因，那我们在使用的时候就可以根据需要调用不同参数的方法了；既然我们看了加锁的源码，那就顺着分析下加锁，续期看门狗的核心逻辑。

tryLockInnerAsync异步加锁任务的核心代码如下：

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { 
       //锁的租约时间 
       internalLockLeaseTime = unit.toMillis(leaseTime);
    						//执行lua脚本，向redis实例发送命令
            return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                 //如果分布式锁lockKey不存在，执行hincrby命令加锁，并设置过期时间 
                      "if (redis.call('exists', KEYS[1]) == 0) then " +
                          "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                          "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                          "return nil; " +
                      "end; " +
                 //如果分布式锁lockKey已经存在，并且value也匹配，表示当前线程持有的锁，那么重入次数加1，并设置过期时间 
                      "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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
        }

*  KEYS\[1\]就是Collections.singletonList(getName())，表示分布式锁的key，即lockKey;
 *  ARGV\[1\]就是internalLockLeaseTime，即锁的租约时间，默认30s;
 *  ARGV\[2\]就是getLockName(threadId)，是获取锁时的唯一值，即UUID+threadId;

加锁成功后会执行一个定时调度任务scheduleExpirationRenewal给锁续期，核心代码如下：：

protected RFuture<Boolean> renewExpirationAsync(long threadId) { 
            return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            //如果分布式锁lockKey和value值都匹配，则给锁续期，并返回1，否则返回0
                    "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return 1; " +
                    "end; " +
                    "return 0;",
                Collections.<Object>singletonList(getName()), 
                internalLockLeaseTime, getLockName(threadId));
        }

*  KEYS\[1\]是就是Collections.singletonList(getName())，表示分布式锁的key，即lockKey;
 *  ARGV\[1\]就是internalLockLeaseTime，即锁的租约时间，默认30s;
 *  ARGV\[2\]就是getLockName(threadId)，是获取锁时的唯一值，即UUID+threadId;

解锁核心方法：

protected RFuture<Boolean> unlockInnerAsync(long threadId) { 
            return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                     //如果分布式锁lockKey存在，但是value不匹配，表示锁已经被占用，那么直接返回null 
                    "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                        "return nil;" +
                    "end; " +
                     //如果当前线程占有分布式锁，那么将重入次数减1 
                    "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                    //如果重入次数减1后大于0，表示分布式锁有重入过，那么只设置失效时间，还不可以删除 
                    "if (counter > 0) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                        "return 0; " +
                    "else " +
                     //重入次数减1后如果为0，表示分布式锁只获取过1次，那么删除这个KEY,并发布解锁消息 
                        "redis.call('del', KEYS[1]); " +
                        "redis.call('publish', KEYS[2], ARGV[1]); " +
                        "return 1; "+
                    "end; " +
                    "return nil;",
                    //这五个参数分别对应KEYS[1],KEYS[2],ARGV[1],ARGV[2],ARGV[3] 
                    Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
    
        }

##### 2.公平锁（Fair Lock） #####

基于Redis的Redisson分布式可重入公平锁也是实现了`java.util.concurrent.locks.Lock`接口的一种`RLock`对象。同时还提供了[异步（Async）][Async]、[反射式（Reactive）][Reactive]和[RxJava2标准][RxJava2]的接口。它保证了当多个Redisson客户端线程同时请求加锁时，优先分配给先发出请求的线程。所有请求线程会在一个队列中排队，当某个线程出现宕机时，Redisson会等待5秒后继续下一个线程，也就是说如果前面有5个线程都处于等待状态，那么后面的线程会等待至少25秒。

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

大家都知道，如果负责储存这个分布式锁的Redis节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改[Config.lockWatchdogTimeout][]来另行指定。

另外Redisson还通过加锁的方法提供了`leaseTime`的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

// 10秒钟以后自动解锁
    // 无需调用unlock方法手动解锁
    fairLock.lock(10, TimeUnit.SECONDS);
    
    // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
    boolean res = fairLock.tryLock(100, 10, TimeUnit.SECONDS);
    ...
    fairLock.unlock();

Redisson同时还为分布式可重入公平锁提供了异步执行的相关方法：

RLock fairLock = redisson.getFairLock("anyLock");
    fairLock.lockAsync();
    fairLock.lockAsync(10, TimeUnit.SECONDS);
    Future<Boolean> res = fairLock.tryLockAsync(100, 10, TimeUnit.SECONDS);

##### 3.联锁(MultiLock) #####

基于Redis的Redisson分布式联锁[`RedissonMultiLock`][RedissonMultiLock]对象可以将多个`RLock`对象关联为一个联锁，每个`RLock`对象实例可以来自于不同的Redisson实例。

RLock lock1 = redissonInstance1.getLock("lock1");
    RLock lock2 = redissonInstance2.getLock("lock2");
    RLock lock3 = redissonInstance3.getLock("lock3");
    
    RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
    // 同时加锁：lock1 lock2 lock3
    // 所有的锁都上锁成功才算成功。
    lock.lock();
    ...
    lock.unlock();

大家都知道，如果负责储存某些分布式锁的某些Redis节点宕机以后，而且这些锁正好处于锁住的状态时，这些锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改[Config.lockWatchdogTimeout][]来另行指定。

另外Redisson还通过加锁的方法提供了`leaseTime`的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
    // 给lock1，lock2，lock3加锁，如果没有手动解开的话，10秒钟后将会自动解开
    lock.lock(10, TimeUnit.SECONDS);
    
    // 为加锁等待100秒时间，并在加锁成功10秒钟后自动解开
    boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
    ...
    lock.unlock();

##### 4.红锁（RedLock） #####

基于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();

另外Redisson还通过加锁的方法提供了`leaseTime`的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
    // 给lock1，lock2，lock3加锁，如果没有手动解开的话，10秒钟后将会自动解开
    lock.lock(10, TimeUnit.SECONDS);
    
    // 为加锁等待100秒时间，并在加锁成功10秒钟后自动解开
    boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
    ...
    lock.unlock();

##### 5.读写锁（ReadWriteLock） #####

基于Redis的Redisson分布式可重入读写锁[`RReadWriteLock`][RReadWriteLock] Java对象实现了`java.util.concurrent.locks.ReadWriteLock`接口。其中读锁和写锁都继承了[RLock][]接口。

分布式可重入读写锁允许同时有多个读锁和一个写锁处于加锁状态。

RReadWriteLock rwlock = redisson.getReadWriteLock("anyRWLock");
    // 最常见的使用方法
    rwlock.readLock().lock();
    // 或
    rwlock.writeLock().lock();

另外Redisson还通过加锁的方法提供了`leaseTime`的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

// 10秒钟以后自动解锁
    // 无需调用unlock方法手动解锁
    rwlock.readLock().lock(10, TimeUnit.SECONDS);
    // 或
    rwlock.writeLock().lock(10, TimeUnit.SECONDS);
    
    // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
    boolean res = rwlock.readLock().tryLock(100, 10, TimeUnit.SECONDS);
    // 或
    boolean res = rwlock.writeLock().tryLock(100, 10, TimeUnit.SECONDS);
    ...
    lock.unlock();

##### 6.信号量（Semaphore） #####

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

RSemaphore semaphore = redisson.getSemaphore("semaphore");
    semaphore.acquire();
    //或
    semaphore.acquireAsync();
    semaphore.acquire(23);
    semaphore.tryAcquire();
    //或
    semaphore.tryAcquireAsync();
    semaphore.tryAcquire(23, TimeUnit.SECONDS);
    //或
    semaphore.tryAcquireAsync(23, TimeUnit.SECONDS);
    semaphore.release(10);
    semaphore.release();
    //或
    semaphore.releaseAsync();

##### 7.可过期性信号量（PermitExpirableSemaphore） #####

基于Redis的Redisson可过期性信号量（[PermitExpirableSemaphore][]）是在`RSemaphore`对象的基础上，为每个信号增加了一个过期时间。每个信号可以通过独立的ID来辨识，释放时只能通过提交这个ID才能释放。它提供了[异步（Async）][Async 2]、[反射式（Reactive）][Reactive 2]和[RxJava2标准][RxJava2 2]的接口。

RPermitExpirableSemaphore semaphore = redisson.getPermitExpirableSemaphore("mySemaphore");
    String permitId = semaphore.acquire();
    // 获取一个信号，有效期只有2秒钟。
    String permitId = semaphore.acquire(2, TimeUnit.SECONDS);
    // ...
    semaphore.release(permitId);

##### 8.闭锁（CountDownLatch） #####

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

RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
    latch.trySetCount(1);
    latch.await();
    
    // 在其他线程或其他JVM里
    RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
    latch.countDown();

参考：[https://github.com/redisson/redisson/wiki/8.-%E5%88%86%E5%B8%83%E5%BC%8F%E9%94%81%E5%92%8C%E5%90%8C%E6%AD%A5%E5%99%A8][https_github.com_redisson_redisson_wiki_8.-_E5_88_86_E5_B8_83_E5_BC_8F_E9_94_81_E5_92_8C_E5_90_8C_E6_AD_A5_E5_99_A8]

GitHub地址：[https://github.com/mingyang66/spring-parent/tree/master/sgrain-spring-boot-redis-service/doc][https_github.com_mingyang66_spring-parent_tree_master_sgrain-spring-boot-redis-service_doc]

[Config.lockWatchdogTimeout]: https://github.com/redisson/redisson/wiki/2.-%E9%85%8D%E7%BD%AE%E6%96%B9%E6%B3%95#lockwatchdogtimeout%E7%9B%91%E6%8E%A7%E9%94%81%E7%9A%84%E7%9C%8B%E9%97%A8%E7%8B%97%E8%B6%85%E6%97%B6%E5%8D%95%E4%BD%8D%E6%AF%AB%E7%A7%92
[Semaphore]: https://github.com/redisson/redisson/wiki/8.-%E5%88%86%E5%B8%83%E5%BC%8F%E9%94%81%E5%92%8C%E5%90%8C%E6%AD%A5%E5%99%A8#86-%E4%BF%A1%E5%8F%B7%E9%87%8Fsemaphore
[Async]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RLockAsync.html
[Reactive]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RLockReactive.html
[RxJava2]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RLockRx.html
[RedissonMultiLock]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/RedissonMultiLock.html
[Redlock]: http://redis.cn/topics/distlock.html
[RReadWriteLock]: http://static.javadoc.io/org.redisson/redisson/3.4.3/org/redisson/api/RReadWriteLock.html
[RLock]: https://github.com/redisson/redisson/wiki/8.-%E5%88%86%E5%B8%83%E5%BC%8F%E9%94%81%E5%92%8C%E5%90%8C%E6%AD%A5%E5%99%A8#81-%E5%8F%AF%E9%87%8D%E5%85%A5%E9%94%81reentrant-lock
[Semaphore 1]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RSemaphore.html
[Async 1]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RSemaphoreAsync.html
[Reactive 1]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RSemaphoreReactive.html
[RxJava2 1]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RSemaphoreRx.html
[PermitExpirableSemaphore]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RPermitExpirableSemaphore.html
[Async 2]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RPermitExpirableSemaphoreAsync.html
[Reactive 2]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RPermitExpirableSemaphoreReactive.html
[RxJava2 2]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RPermitExpirableSemaphoreRx.html
[CountDownLatch]: http://static.javadoc.io/org.redisson/redisson/3.10.0/org/redisson/api/RCountDownLatch.html
[https_github.com_redisson_redisson_wiki_8.-_E5_88_86_E5_B8_83_E5_BC_8F_E9_94_81_E5_92_8C_E5_90_8C_E6_AD_A5_E5_99_A8]: https://github.com/redisson/redisson/wiki/8.-%E5%88%86%E5%B8%83%E5%BC%8F%E9%94%81%E5%92%8C%E5%90%8C%E6%AD%A5%E5%99%A8
[https_github.com_mingyang66_spring-parent_tree_master_sgrain-spring-boot-redis-service_doc]: https://github.com/mingyang66/spring-parent/tree/master/sgrain-spring-boot-redis-service/doc