分布式锁—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使用演示

public class RedissonDemo {     public static void main(String[] args) throws Exception {         //连接3主3从的Redis CLuster         Config config = new Config();         ...         //Semaphore         RedissonClient redisson = Redisson.create(config);         final RSemaphore semaphore = redisson.getSemaphore("semaphore");         semaphore.trySetPermits(3);         for (int i = 0; i < 10; i++) {             new Thread(new Runnable() {                 public void run() {                     try {                         System.out.println(new Date() + "：线程[" + Thread.currentThread().getName() + "]尝试获取Semaphore锁");                         semaphore.acquire();                         System.out.println(new Date() + "：线程[" + Thread.currentThread().getName() + "]成功获取到了Semaphore锁，开始工作");                         Thread.sleep(3000);                         semaphore.release();                         System.out.println(new Date() + "：线程[" + Thread.currentThread().getName() + "]释放Semaphore锁");                     } catch (Exception e) {                         e.printStackTrace();                     }                 }             }).start();         }     } }

3.Redisson的Semaphore源码剖析

(1)Semaphore的初始化

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

(3)客户端尝试获取Semaphore的锁

(4)客户端释放Semaphore的锁

(1)Semaphore的初始化

public class Redisson implements RedissonClient {     //Redis的连接管理器，封装了一个Config实例     protected final ConnectionManager connectionManager;     //Redis的命令执行器，封装了一个ConnectionManager实例     protected final CommandAsyncExecutor commandExecutor;     ...     protected Redisson(Config config) {         this.config = config;         Config configCopy = new Config(config);         //初始化Redis的连接管理器         connectionManager = ConfigSupport.createConnectionManager(configCopy);         ...           //初始化Redis的命令执行器         commandExecutor = new CommandSyncService(connectionManager, objectBuilder);         ...     }      @Override     public RSemaphore getSemaphore(String name) {         return new RedissonSemaphore(commandExecutor, name);     }     ... }  public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {     private final SemaphorePubSub semaphorePubSub;     final CommandAsyncExecutor commandExecutor;      public RedissonSemaphore(CommandAsyncExecutor commandExecutor, String name) {         super(commandExecutor, name);         this.commandExecutor = commandExecutor;         this.semaphorePubSub = commandExecutor.getConnectionManager().getSubscribeService().getSemaphorePubSub();     }     ... }

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

public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {     ...     @Override     public boolean trySetPermits(int permits) {         return get(trySetPermitsAsync(permits));     }      @Override     public RFuture<Boolean> trySetPermitsAsync(int permits) {         RFuture<Boolean> future = commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,             //执行命令"get semaphore"，获取到当前的数值             "local value = redis.call('get', KEYS[1]); " +             "if (value == false) then " +                 //然后执行命令"set semaphore 3"                 //设置这个信号量允许客户端同时获取锁的总数量为3                 "redis.call('set', KEYS[1], ARGV[1]); " +                 "redis.call('publish', KEYS[2], ARGV[1]); " +                 "return 1;" +             "end;" +             "return 0;",             Arrays.asList(getRawName(), getChannelName()),             permits         );          if (log.isDebugEnabled()) {             future.onComplete((r, e) -> {                 if (r) {                     log.debug("permits set, permits: {}, name: {}", permits, getName());                 } else {                     log.debug("unable to set permits, permits: {}, name: {}", permits, getName());                 }             });         }         return future;     }     ... }

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

(3)客户端尝试获取Semaphore的锁

public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {     ...     private final SemaphorePubSub semaphorePubSub;     final CommandAsyncExecutor commandExecutor;      public RedissonSemaphore(CommandAsyncExecutor commandExecutor, String name) {         super(commandExecutor, name);         this.commandExecutor = commandExecutor;         this.semaphorePubSub = commandExecutor.getConnectionManager().getSubscribeService().getSemaphorePubSub();     }      @Override     public void acquire() throws InterruptedException {         acquire(1);     }          @Override     public void acquire(int permits) throws InterruptedException {         if (tryAcquire(permits)) {             return;         }         CompletableFuture<RedissonLockEntry> future = subscribe();         commandExecutor.syncSubscriptionInterrupted(future);         try {             while (true) {                 if (tryAcquire(permits)) {                     return;                 }                 //获取Redisson的Semaphore失败，于是便调用本地JDK的Semaphore的acquire()方法，此时当前线程会被阻塞                 //之后如果Redisson的Semaphore释放了锁，那么当前客户端便会通过监听订阅事件释放本地JDK的Semaphore，唤醒被阻塞的线程，继续执行while循环                 //注意：getLatch()返回的是JDK的Semaphore = "new Semaphore(0)" ==> (state - permits)                   //首先调用CommandAsyncService.getNow()方法                 //然后调用RedissonLockEntry.getLatch()方法                 //接着调用JDK的Semaphore的acquire()方法                 commandExecutor.getNow(future).getLatch().acquire();             }         } finally {             unsubscribe(commandExecutor.getNow(future));         }     }          @Override     public boolean tryAcquire(int permits) {         //异步转同步         return get(tryAcquireAsync(permits));     }          @Override     public RFuture<Boolean> tryAcquireAsync(int permits) {         if (permits < 0) {             throw new IllegalArgumentException("Permits amount can't be negative");         }         if (permits == 0) {             return RedissonPromise.newSucceededFuture(true);         }         return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,             //执行命令"get semaphore"，获取到当前值             "local value = redis.call('get', KEYS[1]); "+             //如果semaphore的当前值不是false，且大于客户端线程申请获取锁的数量             "if (value ~= false and tonumber(value) >= tonumber(ARGV[1])) then " +                 //执行"decrby semaphore 1"，将信号量允许获取锁的总数量递减1                 "local val = redis.call('decrby', KEYS[1], ARGV[1]); " +                 "return 1; " +             "end; " +             //如果semaphore的值变为0，那么客户端就无法获取锁了，此时返回false             "return 0;",             Collections.<Object>singletonList(getRawName()),             permits//ARGV[1]默认是1         );     }     ... }  public class CommandAsyncService implements CommandAsyncExecutor {     ...     @Override     public <V> V getNow(CompletableFuture<V> future) {         try {             return future.getNow(null);         } catch (Exception e) {             return null;         }     }     ... }  public class RedissonLockEntry implements PubSubEntry<RedissonLockEntry> {     private final Semaphore latch;     ...     public RedissonLockEntry(CompletableFuture<RedissonLockEntry> promise) {         super();         this.latch = new Semaphore(0);         this.promise = promise;     }          public Semaphore getLatch() {         return latch;     }     ... }

执行命令"get semaphore"，获取到semaphore的当前值。如果semaphore的当前值不是false，且大于客户端线程申请获取锁的数量。那么就执行"decrby semaphore 1"，将信号量允许获取锁的总数量递减1。

如果semaphore的值变为0，那么客户端就无法获取锁了，此时tryAcquire()方法返回false。表示获取semaphore的锁失败了，于是当前客户端线程便会通过本地JDK的Semaphore进行阻塞。

当客户端后续收到一个订阅事件把本地JDK的Semaphore进行释放后，便会唤醒阻塞线程继续while循环。在while循环中，会不断尝试获取这个semaphore的锁，如此循环往复，直到成功获取。

(4)客户端释放Semaphore的锁

public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {     ...     @Override     public void release() {         release(1);     }      @Override     public void release(int permits) {         get(releaseAsync(permits));     }      @Override     public RFuture<Void> releaseAsync(int permits) {         if (permits < 0) {             throw new IllegalArgumentException("Permits amount can't be negative");         }         if (permits == 0) {             return RedissonPromise.newSucceededFuture(null);         }          RFuture<Void> future = commandExecutor.evalWriteAsync(getRawName(), StringCodec.INSTANCE, RedisCommands.EVAL_VOID,             //执行命令"incrby semaphore 1"             "local value = redis.call('incrby', KEYS[1], ARGV[1]); " +             "redis.call('publish', KEYS[2], value); ",             Arrays.asList(getRawName(), getChannelName()),             permits         );         if (log.isDebugEnabled()) {             future.onComplete((o, e) -> {                 if (e == null) {                     log.debug("released, permits: {}, name: {}", permits, getName());                 }             });         }         return future;     }     ... }  //订阅semaphore不为0的事件，semaphore不为0时会触发执行这里的监听回调 public class SemaphorePubSub extends PublishSubscribe<RedissonLockEntry> {     public SemaphorePubSub(PublishSubscribeService service) {         super(service);     }          @Override     protected RedissonLockEntry createEntry(CompletableFuture<RedissonLockEntry> newPromise) {         return new RedissonLockEntry(newPromise);     }          @Override     protected void onMessage(RedissonLockEntry value, Long message) {     Runnable runnableToExecute = value.getListeners().poll();         if (runnableToExecute != null) {             runnableToExecute.run();         }         //将客户端本地JDK的Semaphore进行释放         value.getLatch().release(Math.min(value.acquired(), message.intValue()));     } }  //订阅锁被释放的事件，锁被释放为0时会触发执行这里的监听回调 public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {     public static final Long UNLOCK_MESSAGE = 0L;     public static final Long READ_UNLOCK_MESSAGE = 1L;          public LockPubSub(PublishSubscribeService service) {         super(service);     }              @Override     protected RedissonLockEntry createEntry(CompletableFuture<RedissonLockEntry> newPromise) {         return new RedissonLockEntry(newPromise);     }          @Override     protected void onMessage(RedissonLockEntry value, Long message) {         if (message.equals(UNLOCK_MESSAGE)) {             Runnable runnableToExecute = value.getListeners().poll();             if (runnableToExecute != null) {                 runnableToExecute.run();             }             value.getLatch().release();         } else if (message.equals(READ_UNLOCK_MESSAGE)) {             while (true) {                 Runnable runnableToExecute = value.getListeners().poll();                 if (runnableToExecute == null) {                     break;                 }                 runnableToExecute.run();             }             //将客户端本地JDK的Semaphore进行释放             value.getLatch().release(value.getLatch().getQueueLength());         }     } }

客户端释放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的线程阻塞。