分布式锁—6.Redisson的同步器组件

大纲
1.Redisson的分布式锁简单总结
2.Redisson的Semaphore简介
3.Redisson的Semaphore源码剖析
4.Redisson的CountDownLatch简介
5.Redisson的CountDownLatch源码剖析
 
1.Redisson的分布式锁简单总结
(1)可重入锁RedissonLock
(2)公平锁RedissonFairLock
(3)联锁MultiLock
(4)红锁RedLock
(5)读写锁之读锁RedissonReadLock和写锁RedissonWriteLock
 
Redisson分布式锁包括：可重入锁、公平锁、联锁、红锁、读写锁。
 
(1)可重入锁RedissonLock
非公平锁，最基础的分布式锁，最常用的锁。
 
(2)公平锁RedissonFairLock
各个客户端尝试获取锁时会排队，按照队列的顺序先后获取锁。
 
(3)联锁MultiLock
可以一次性加多把锁，从而实现一次性锁多个资源。
 
(4)红锁RedLock
RedLock相当于一把锁。虽然利用了MultiLock包裹了多个小锁，但这些小锁并不对应多个资源，而是每个小锁的key对应一个Redis实例。只要大多数的Redis实例加锁成功，就可以认为RedLock加锁成功。RedLock的健壮性要比其他普通锁要好。
 
但是RedLock也有一些场景无法保证正确性，当然RedLock只要求部署主库。比如客户端A尝试向5个Master实例加锁，但仅仅在3个Maste中加锁成功。不幸的是此时3个Master中有1个Master突然宕机了，而且锁key还没同步到该宕机Master的Slave上，此时Salve切换为Master。于是在这5个Master中，由于其中有一个是新切换过来的Master，所以只有2个Master是有客户端A加锁的数据，另外3个Master是没有锁的。但继续不幸的是，此时客户端B来加锁，那么客户端B就很有可能成功在没有锁数据的3个Master上加到锁，从而满足了过半数加锁的要求，最后也完成了加锁，依然发生重复加锁。
 
(5)读写锁之读锁RedissonReadLock和写锁RedissonWriteLock
不同客户端线程的四种加锁情况：
情况一：先加读锁再加读锁，不互斥
情况二：先加读锁再加写锁，互斥
情况三：先加写锁再加读锁，互斥
情况四：先加写锁再加写锁，互斥
 
同一个客户端线程的四种加锁情况：
情况一：先加读锁再加读锁，不互斥
情况二：先加读锁再加写锁，互斥
情况三：先加写锁再加读锁，不互斥
情况四：先加写锁再加写锁，不互斥
 
2.Redisson的Semaphore简介
(1)Redisson的Semaphore原理图
Semaphore也是Redisson支持的一种同步组件。Semaphore作为一个锁机制，可以允许多个线程同时获取一把锁。任何一个线程释放锁之后，其他等待的线程就可以尝试继续获取锁。

(2)Redisson的Semaphore使用演示

1. public class RedissonDemo {
2.     public static void main(String[] args) throws Exception {
3.         //连接3主3从的Redis CLuster
4.         Config config = new Config();
5.         ...
6.         //Semaphore
7.         RedissonClient redisson = Redisson.create(config);
8.         final RSemaphore semaphore = redisson.getSemaphore("semaphore");
9.         semaphore.trySetPermits(3);
10.         for (int i = 0; i < 10; i++) {
11.             new Thread(new Runnable() {
12.                 public void run() {
13.                     try {
14.                         System.out.println(new Date() + "：线程[" + Thread.currentThread().getName() + "]尝试获取Semaphore锁");
15.                         semaphore.acquire();
16.                         System.out.println(new Date() + "：线程[" + Thread.currentThread().getName() + "]成功获取到了Semaphore锁，开始工作");
17.                         Thread.sleep(3000);
18.                         semaphore.release();
19.                         System.out.println(new Date() + "：线程[" + Thread.currentThread().getName() + "]释放Semaphore锁");
20.                     } catch (Exception e) {
21.                         e.printStackTrace();
22.                     }
23.                 }
24.             }).start();
25.         }
26.     }
27. }

复制代码

 
3.Redisson的Semaphore源码剖析
(1)Semaphore的初始化
(2)Semaphore设置允许获取的锁数量
(3)客户端尝试获取Semaphore的锁
(4)客户端释放Semaphore的锁
 
(1)Semaphore的初始化

1. public class Redisson implements RedissonClient {
2.     //Redis的连接管理器，封装了一个Config实例
3.     protected final ConnectionManager connectionManager;
4.     //Redis的命令执行器，封装了一个ConnectionManager实例
5.     protected final CommandAsyncExecutor commandExecutor;
6.     ...
7.     protected Redisson(Config config) {
8.         this.config = config;
9.         Config configCopy = new Config(config);
10.         //初始化Redis的连接管理器
11.         connectionManager = ConfigSupport.createConnectionManager(configCopy);
12.         ...  
13.         //初始化Redis的命令执行器
14.         commandExecutor = new CommandSyncService(connectionManager, objectBuilder);
15.         ...
16.     }
18.     @Override
19.     public RSemaphore getSemaphore(String name) {
20.         return new RedissonSemaphore(commandExecutor, name);
21.     }
22.     ...
23. }
25. public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {
26.     private final SemaphorePubSub semaphorePubSub;
27.     final CommandAsyncExecutor commandExecutor;
29.     public RedissonSemaphore(CommandAsyncExecutor commandExecutor, String name) {
30.         super(commandExecutor, name);
31.         this.commandExecutor = commandExecutor;
32.         this.semaphorePubSub = commandExecutor.getConnectionManager().getSubscribeService().getSemaphorePubSub();
33.     }
34.     ...
35. }

复制代码

(2)Semaphore设置允许获取的锁数量

1. public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {
2.     ...
3.     @Override
4.     public boolean trySetPermits(int permits) {
5.         return get(trySetPermitsAsync(permits));
6.     }
8.     @Override
9.     public RFuture<Boolean> trySetPermitsAsync(int permits) {
10.         RFuture<Boolean> future = commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
11.             //执行命令"get semaphore"，获取到当前的数值
12.             "local value = redis.call('get', KEYS[1]); " +
13.             "if (value == false) then " +
14.                 //然后执行命令"set semaphore 3"
15.                 //设置这个信号量允许客户端同时获取锁的总数量为3
16.                 "redis.call('set', KEYS[1], ARGV[1]); " +
17.                 "redis.call('publish', KEYS[2], ARGV[1]); " +
18.                 "return 1;" +
19.             "end;" +
20.             "return 0;",
21.             Arrays.asList(getRawName(), getChannelName()),
22.             permits
23.         );
25.         if (log.isDebugEnabled()) {
26.             future.onComplete((r, e) -> {
27.                 if (r) {
28.                     log.debug("permits set, permits: {}, name: {}", permits, getName());
29.                 } else {
30.                     log.debug("unable to set permits, permits: {}, name: {}", permits, getName());
31.                 }
32.             });
33.         }
34.         return future;
35.     }
36.     ...
37. }

复制代码

首先执行命令"get semaphore"，获取到当前的数值。然后执行命令"set semaphore 3"，也就是设置这个信号量允许客户端同时获取锁的总数量为3。
 
(3)客户端尝试获取Semaphore的锁

1. public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {
2.     ...
3.     private final SemaphorePubSub semaphorePubSub;
4.     final CommandAsyncExecutor commandExecutor;
6.     public RedissonSemaphore(CommandAsyncExecutor commandExecutor, String name) {
7.         super(commandExecutor, name);
8.         this.commandExecutor = commandExecutor;
9.         this.semaphorePubSub = commandExecutor.getConnectionManager().getSubscribeService().getSemaphorePubSub();
10.     }
12.     @Override
13.     public void acquire() throws InterruptedException {
14.         acquire(1);
15.     }
16.    
17.     @Override
18.     public void acquire(int permits) throws InterruptedException {
19.         if (tryAcquire(permits)) {
20.             return;
21.         }
22.         CompletableFuture<RedissonLockEntry> future = subscribe();
23.         commandExecutor.syncSubscriptionInterrupted(future);
24.         try {
25.             while (true) {
26.                 if (tryAcquire(permits)) {
27.                     return;
28.                 }
29.                 //获取Redisson的Semaphore失败，于是便调用本地JDK的Semaphore的acquire()方法，此时当前线程会被阻塞
30.                 //之后如果Redisson的Semaphore释放了锁，那么当前客户端便会通过监听订阅事件释放本地JDK的Semaphore，唤醒被阻塞的线程，继续执行while循环
31.                 //注意：getLatch()返回的是JDK的Semaphore = "new Semaphore(0)" ==> (state - permits)
34.                 //首先调用CommandAsyncService.getNow()方法
35.                 //然后调用RedissonLockEntry.getLatch()方法
36.                 //接着调用JDK的Semaphore的acquire()方法
37.                 commandExecutor.getNow(future).getLatch().acquire();
38.             }
39.         } finally {
40.             unsubscribe(commandExecutor.getNow(future));
41.         }
42.     }
43.    
44.     @Override
45.     public boolean tryAcquire(int permits) {
46.         //异步转同步
47.         return get(tryAcquireAsync(permits));
48.     }
49.    
50.     @Override
51.     public RFuture<Boolean> tryAcquireAsync(int permits) {
52.         if (permits < 0) {
53.             throw new IllegalArgumentException("Permits amount can't be negative");
54.         }
55.         if (permits == 0) {
56.             return RedissonPromise.newSucceededFuture(true);
57.         }
58.         return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
59.             //执行命令"get semaphore"，获取到当前值
60.             "local value = redis.call('get', KEYS[1]); "+
61.             //如果semaphore的当前值不是false，且大于客户端线程申请获取锁的数量
62.             "if (value ~= false and tonumber(value) >= tonumber(ARGV[1])) then " +
63.                 //执行"decrby semaphore 1"，将信号量允许获取锁的总数量递减1
64.                 "local val = redis.call('decrby', KEYS[1], ARGV[1]); " +
65.                 "return 1; " +
66.             "end; " +
67.             //如果semaphore的值变为0，那么客户端就无法获取锁了，此时返回false
68.             "return 0;",
69.             Collections.<Object>singletonList(getRawName()),
70.             permits//ARGV[1]默认是1
71.         );
72.     }
73.     ...
74. }
76. public class CommandAsyncService implements CommandAsyncExecutor {
77.     ...
78.     @Override
79.     public <V> V getNow(CompletableFuture<V> future) {
80.         try {
81.             return future.getNow(null);
82.         } catch (Exception e) {
83.             return null;
84.         }
85.     }
86.     ...
87. }
89. public class RedissonLockEntry implements PubSubEntry<RedissonLockEntry> {
90.     private final Semaphore latch;
91.     ...
92.     public RedissonLockEntry(CompletableFuture<RedissonLockEntry> promise) {
93.         super();
94.         this.latch = new Semaphore(0);
95.         this.promise = promise;
96.     }
97.    
98.     public Semaphore getLatch() {
99.         return latch;
100.     }
101.     ...
102. }

复制代码

执行命令"get semaphore"，获取到semaphore的当前值。如果semaphore的当前值不是false，且大于客户端线程申请获取锁的数量。那么就执行"decrby semaphore 1"，将信号量允许获取锁的总数量递减1。
 
如果semaphore的值变为0，那么客户端就无法获取锁了，此时tryAcquire()方法返回false。表示获取semaphore的锁失败了，于是当前客户端线程便会通过本地JDK的Semaphore进行阻塞。
 
当客户端后续收到一个订阅事件把本地JDK的Semaphore进行释放后，便会唤醒阻塞线程继续while循环。在while循环中，会不断尝试获取这个semaphore的锁，如此循环往复，直到成功获取。
 
(4)客户端释放Semaphore的锁

1. public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {
2.     ...
3.     @Override
4.     public void release() {
5.         release(1);
6.     }
8.     @Override
9.     public void release(int permits) {
10.         get(releaseAsync(permits));
11.     }
13.     @Override
14.     public RFuture<Void> releaseAsync(int permits) {
15.         if (permits < 0) {
16.             throw new IllegalArgumentException("Permits amount can't be negative");
17.         }
18.         if (permits == 0) {
19.             return RedissonPromise.newSucceededFuture(null);
20.         }
22.         RFuture<Void> future = commandExecutor.evalWriteAsync(getRawName(), StringCodec.INSTANCE, RedisCommands.EVAL_VOID,
23.             //执行命令"incrby semaphore 1"
24.             "local value = redis.call('incrby', KEYS[1], ARGV[1]); " +
25.             "redis.call('publish', KEYS[2], value); ",
26.             Arrays.asList(getRawName(), getChannelName()),
27.             permits
28.         );
29.         if (log.isDebugEnabled()) {
30.             future.onComplete((o, e) -> {
31.                 if (e == null) {
32.                     log.debug("released, permits: {}, name: {}", permits, getName());
33.                 }
34.             });
35.         }
36.         return future;
37.     }
38.     ...
39. }
41. //订阅semaphore不为0的事件，semaphore不为0时会触发执行这里的监听回调
42. public class SemaphorePubSub extends PublishSubscribe<RedissonLockEntry> {
43.     public SemaphorePubSub(PublishSubscribeService service) {
44.         super(service);
45.     }
46.    
47.     @Override
48.     protected RedissonLockEntry createEntry(CompletableFuture<RedissonLockEntry> newPromise) {
49.         return new RedissonLockEntry(newPromise);
50.     }
51.    
52.     @Override
53.     protected void onMessage(RedissonLockEntry value, Long message) {
54.     Runnable runnableToExecute = value.getListeners().poll();
55.         if (runnableToExecute != null) {
56.             runnableToExecute.run();
57.         }
58.         //将客户端本地JDK的Semaphore进行释放
59.         value.getLatch().release(Math.min(value.acquired(), message.intValue()));
60.     }
61. }
63. //订阅锁被释放的事件，锁被释放为0时会触发执行这里的监听回调
64. public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {
65.     public static final Long UNLOCK_MESSAGE = 0L;
66.     public static final Long READ_UNLOCK_MESSAGE = 1L;
67.    
68.     public LockPubSub(PublishSubscribeService service) {
69.         super(service);
70.     }  
71.       
72.     @Override
73.     protected RedissonLockEntry createEntry(CompletableFuture<RedissonLockEntry> newPromise) {
74.         return new RedissonLockEntry(newPromise);
75.     }
76.    
77.     @Override
78.     protected void onMessage(RedissonLockEntry value, Long message) {
79.         if (message.equals(UNLOCK_MESSAGE)) {
80.             Runnable runnableToExecute = value.getListeners().poll();
81.             if (runnableToExecute != null) {
82.                 runnableToExecute.run();
83.             }
84.             value.getLatch().release();
85.         } else if (message.equals(READ_UNLOCK_MESSAGE)) {
86.             while (true) {
87.                 Runnable runnableToExecute = value.getListeners().poll();
88.                 if (runnableToExecute == null) {
89.                     break;
90.                 }
91.                 runnableToExecute.run();
92.             }
93.             //将客户端本地JDK的Semaphore进行释放
94.             value.getLatch().release(value.getLatch().getQueueLength());
95.         }
96.     }
97. }

复制代码

客户端释放Semaphore的锁时，会执行命令"incrby semaphore 1"。每当客户端释放掉permits个锁，就会将信号量的值累加permits，这样Semaphore信号量的值就不再是0了。然后通过publish命令发布一个事件，之后订阅了该事件的其他客户端都会对getLatch()返回的本地JDK的Semaphore进行加1。于是其他客户端正在被本地JDK的Semaphore进行阻塞的线程，就会被唤醒继续执行。此时其他客户端就可以尝试获取到这个信号量的锁，然后再次将这个Semaphore的值递减1。
 
4.Redisson的CountDownLatch简介
(1)Redisson的CountDownLatch原理图解
(2)Redisson的CountDownLatch使用演示
 
(1)Redisson的CountDownLatch原理图解
CountDownLatch的基本原理：要求必须有n个线程来进行countDown，才能让执行await的线程继续执行。如果没有达到指定数量的线程来countDown，会导致执行await的线程阻塞。

(2)Redisson的CountDownLatch使用演示

1. public class RedissonDemo {
2.     public static void main(String[] args) throws Exception {
3.         //连接3主3从的Redis CLuster
4.         Config config = new Config();
5.         ...
6.         //CountDownLatch
7.         final RedissonClient redisson = Redisson.create(config);
8.         RCountDownLatch latch = redisson.getCountDownLatch("myCountDownLatch");
9.         //1.设置可以countDown的数量为3
10.         latch.trySetCount(3);
11.         System.out.println(new Date() + "：线程[" + Thread.currentThread().getName() + "]设置了必须有3个线程执行countDown，进入等待中。。。");
13.         for (int i = 0; i < 3; i++) {
14.             new Thread(new Runnable() {
15.                 public void run() {
16.                     try {
17.                         System.out.println(new Date() + "：线程[" + Thread.currentThread().getName() + "]在做一些操作，请耐心等待。。。。。。");
18.                         Thread.sleep(3000);
19.                         RCountDownLatch localLatch = redisson.getCountDownLatch("myCountDownLatch");
20.                         localLatch.countDown();
21.                         System.out.println(new Date() + "：线程[" + Thread.currentThread().getName() + "]执行countDown操作");
22.                     } catch (Exception e) {
23.                         e.printStackTrace();
24.                     }
25.                 }
26.             }).start();
27.         }
28.         latch.await();
29.         System.out.println(new Date() + "：线程[" + Thread.currentThread().getName() + "]收到通知，有3个线程都执行了countDown操作，可以继续往下执行");
30.     }
31. }

复制代码

 
5.Redisson的CountDownLatch源码剖析
(1)CountDownLatch的初始化
(2)trySetCount()方法设置countDown的数量
(3)awati()方法进行阻塞等待
(4)countDown()方法对countDown的数量递减
 
(1)CountDownLatch的初始化

1. public class Redisson implements RedissonClient {
2.     //Redis的连接管理器，封装了一个Config实例
3.     protected final ConnectionManager connectionManager;
4.     //Redis的命令执行器，封装了一个ConnectionManager实例
5.     protected final CommandAsyncExecutor commandExecutor;
6.     ...
7.     protected Redisson(Config config) {
8.         this.config = config;
9.         Config configCopy = new Config(config);
10.         //初始化Redis的连接管理器
11.         connectionManager = ConfigSupport.createConnectionManager(configCopy);
12.         ...  
13.         //初始化Redis的命令执行器
14.         commandExecutor = new CommandSyncService(connectionManager, objectBuilder);
15.         ...
16.     }
18.     @Override
19.     public RCountDownLatch getCountDownLatch(String name) {
20.         return new RedissonCountDownLatch(commandExecutor, name);
21.     }
22.     ...
23. }
25. public class RedissonCountDownLatch extends RedissonObject implements RCountDownLatch {
26.     ...
27.     private final CountDownLatchPubSub pubSub;
28.     private final String id;
29.     protected RedissonCountDownLatch(CommandAsyncExecutor commandExecutor, String name) {
30.         super(commandExecutor, name);
31.         this.id = commandExecutor.getConnectionManager().getId();
32.         this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getCountDownLatchPubSub();
33.     }
34.     ...
35. }

复制代码

(2)trySetCount()方法设置countDown的数量
trySetCount()方法的工作就是执行命令"set myCountDownLatch 3"。

1. public class RedissonCountDownLatch extends RedissonObject implements RCountDownLatch {
2.     ...
3.     @Override
4.     public boolean trySetCount(long count) {
5.         return get(trySetCountAsync(count));
6.     }
8.     @Override
9.     public RFuture<Boolean> trySetCountAsync(long count) {
10.         return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
11.             "if redis.call('exists', KEYS[1]) == 0 then " +
12.                 "redis.call('set', KEYS[1], ARGV[2]); " +
13.                 "redis.call('publish', KEYS[2], ARGV[1]); " +
14.                 "return 1 " +
15.             "else " +
16.                 "return 0 " +
17.             "end",
18.             Arrays.asList(getRawName(), getChannelName()),
19.             CountDownLatchPubSub.NEW_COUNT_MESSAGE,
20.             count
21.         );
22.     }
23.     ...
24. }

复制代码

(3)awati()方法进行阻塞等待

1. public class RedissonCountDownLatch extends RedissonObject implements RCountDownLatch {
2.     ...
3.     @Override
4.     public void await() throws InterruptedException {
5.         if (getCount() == 0) {
6.             return;
7.         }
8.         CompletableFuture<RedissonCountDownLatchEntry> future = subscribe();
9.         try {
10.             commandExecutor.syncSubscriptionInterrupted(future);
11.             while (getCount() > 0) {
12.                 // waiting for open state
13.                 //获取countDown的数量还大于0，就先阻塞线程，然后再等待唤醒，执行while循环
14.                 //其中getLatch()返回的是JDK的semaphore = "new Semaphore(0)" ==> (state - permits)
15.                 commandExecutor.getNow(future).getLatch().await();
16.             }
17.         } finally {
18.             unsubscribe(commandExecutor.getNow(future));
19.         }
20.     }
22.     @Override
23.     public long getCount() {
24.         return get(getCountAsync());
25.     }
27.     @Override
28.     public RFuture<Long> getCountAsync() {
29.         //执行命令"get myCountDownLatch"
30.         return commandExecutor.writeAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.GET_LONG, getRawName());
31.     }
32.     ...
33. }

复制代码

在while循环中，首先会执行命令"get myCountDownLatch"去获取countDown值。如果该值不大于0，就退出循环不阻塞线程。如果该值大于0，则说明还没有指定数量的线程去执行countDown操作，于是就会先阻塞线程，然后再等待唤醒来继续循环。
 
(4)countDown()方法对countDown的数量递减
[code]public class RedissonCountDownLatch extends RedissonObject implements RCountDownLatch {    ...    @Override    public void countDown() {        get(countDownAsync());    }    @Override    public RFuture countDownAsync() {        return commandExecutor.evalWriteNoRetryAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,            "local v = redis.call('decr', KEYS[1]);" +            "if v