找人建设网站,个人网站设计企业,网站架构工程师,专业制作外贸网站的公司大纲
1.Redisson的分布式锁简单总结
2.Redisson的Semaphore简介
3.Redisson的Semaphore源码剖析
4.Redisson的CountDownLatch简介
5.Redisson的CountDownLatch源码剖析 1.Redisson的分布式锁简单总结
(1)可重入锁RedissonLock
(2)公平锁RedissonFairLock
(3)联锁MultiL…大纲
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 CLusterConfig config new Config();...//SemaphoreRedissonClient 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);...}Overridepublic 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 {...Overridepublic boolean trySetPermits(int permits) {return get(trySetPermitsAsync(permits));}Overridepublic RFutureBoolean trySetPermitsAsync(int permits) {RFutureBoolean 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//设置这个信号量允许客户端同时获取锁的总数量为3redis.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();}Overridepublic void acquire() throws InterruptedException {acquire(1);}Overridepublic void acquire(int permits) throws InterruptedException {if (tryAcquire(permits)) {return;}CompletableFutureRedissonLockEntry 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));}}Overridepublic boolean tryAcquire(int permits) {//异步转同步return get(tryAcquireAsync(permits));}Overridepublic RFutureBoolean tryAcquireAsync(int permits) {if (permits 0) {throw new IllegalArgumentException(Permits amount cant 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将信号量允许获取锁的总数量递减1local val redis.call(decrby, KEYS[1], ARGV[1]); return 1; end; //如果semaphore的值变为0那么客户端就无法获取锁了此时返回falsereturn 0;,Collections.ObjectsingletonList(getRawName()),permits//ARGV[1]默认是1);}...
}public class CommandAsyncService implements CommandAsyncExecutor {...Overridepublic V V getNow(CompletableFutureV future) {try {return future.getNow(null);} catch (Exception e) {return null;}}...
}public class RedissonLockEntry implements PubSubEntryRedissonLockEntry {private final Semaphore latch;...public RedissonLockEntry(CompletableFutureRedissonLockEntry 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 {...Overridepublic void release() {release(1);}Overridepublic void release(int permits) {get(releaseAsync(permits));}Overridepublic RFutureVoid releaseAsync(int permits) {if (permits 0) {throw new IllegalArgumentException(Permits amount cant be negative);}if (permits 0) {return RedissonPromise.newSucceededFuture(null);}RFutureVoid future commandExecutor.evalWriteAsync(getRawName(), StringCodec.INSTANCE, RedisCommands.EVAL_VOID,//执行命令incrby semaphore 1local 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 PublishSubscribeRedissonLockEntry {public SemaphorePubSub(PublishSubscribeService service) {super(service);}Overrideprotected RedissonLockEntry createEntry(CompletableFutureRedissonLockEntry newPromise) {return new RedissonLockEntry(newPromise);}Overrideprotected 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 PublishSubscribeRedissonLockEntry {public static final Long UNLOCK_MESSAGE 0L;public static final Long READ_UNLOCK_MESSAGE 1L;public LockPubSub(PublishSubscribeService service) {super(service);} Overrideprotected RedissonLockEntry createEntry(CompletableFutureRedissonLockEntry newPromise) {return new RedissonLockEntry(newPromise);}Overrideprotected 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的线程阻塞。 (2)Redisson的CountDownLatch使用演示
public class RedissonDemo {public static void main(String[] args) throws Exception {//连接3主3从的Redis CLusterConfig config new Config();...//CountDownLatchfinal RedissonClient redisson Redisson.create(config);RCountDownLatch latch redisson.getCountDownLatch(myCountDownLatch);//1.设置可以countDown的数量为3latch.trySetCount(3);System.out.println(new Date() 线程[ Thread.currentThread().getName() ]设置了必须有3个线程执行countDown进入等待中。。。);for (int i 0; i 3; i) {new Thread(new Runnable() {public void run() {try {System.out.println(new Date() 线程[ Thread.currentThread().getName() ]在做一些操作请耐心等待。。。。。。);Thread.sleep(3000);RCountDownLatch localLatch redisson.getCountDownLatch(myCountDownLatch);localLatch.countDown();System.out.println(new Date() 线程[ Thread.currentThread().getName() ]执行countDown操作);} catch (Exception e) {e.printStackTrace();}}}).start();}latch.await();System.out.println(new Date() 线程[ Thread.currentThread().getName() ]收到通知有3个线程都执行了countDown操作可以继续往下执行);}
} 5.Redisson的CountDownLatch源码剖析
(1)CountDownLatch的初始化
(2)trySetCount()方法设置countDown的数量
(3)awati()方法进行阻塞等待
(4)countDown()方法对countDown的数量递减 (1)CountDownLatch的初始化
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);...}Overridepublic RCountDownLatch getCountDownLatch(String name) {return new RedissonCountDownLatch(commandExecutor, name);}...
}public class RedissonCountDownLatch extends RedissonObject implements RCountDownLatch {...private final CountDownLatchPubSub pubSub;private final String id;protected RedissonCountDownLatch(CommandAsyncExecutor commandExecutor, String name) {super(commandExecutor, name);this.id commandExecutor.getConnectionManager().getId();this.pubSub commandExecutor.getConnectionManager().getSubscribeService().getCountDownLatchPubSub();}...
}
(2)trySetCount()方法设置countDown的数量
trySetCount()方法的工作就是执行命令set myCountDownLatch 3。
public class RedissonCountDownLatch extends RedissonObject implements RCountDownLatch {...Overridepublic boolean trySetCount(long count) {return get(trySetCountAsync(count));}Overridepublic RFutureBoolean trySetCountAsync(long count) {return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,if redis.call(exists, KEYS[1]) 0 then redis.call(set, KEYS[1], ARGV[2]); redis.call(publish, KEYS[2], ARGV[1]); return 1 else return 0 end,Arrays.asList(getRawName(), getChannelName()),CountDownLatchPubSub.NEW_COUNT_MESSAGE,count);}...
}
(3)awati()方法进行阻塞等待
public class RedissonCountDownLatch extends RedissonObject implements RCountDownLatch {...Overridepublic void await() throws InterruptedException {if (getCount() 0) {return;}CompletableFutureRedissonCountDownLatchEntry future subscribe();try {commandExecutor.syncSubscriptionInterrupted(future);while (getCount() 0) {// waiting for open state//获取countDown的数量还大于0就先阻塞线程然后再等待唤醒执行while循环//其中getLatch()返回的是JDK的semaphore new Semaphore(0) (state - permits)commandExecutor.getNow(future).getLatch().await();}} finally {unsubscribe(commandExecutor.getNow(future));}}Overridepublic long getCount() {return get(getCountAsync());}Overridepublic RFutureLong getCountAsync() {//执行命令get myCountDownLatchreturn commandExecutor.writeAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.GET_LONG, getRawName());}...
}
在while循环中首先会执行命令get myCountDownLatch去获取countDown值。如果该值不大于0就退出循环不阻塞线程。如果该值大于0则说明还没有指定数量的线程去执行countDown操作于是就会先阻塞线程然后再等待唤醒来继续循环。 (4)countDown()方法对countDown的数量递减
public class RedissonCountDownLatch extends RedissonObject implements RCountDownLatch {...Overridepublic void countDown() {get(countDownAsync());}Overridepublic RFutureVoid countDownAsync() {return commandExecutor.evalWriteNoRetryAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,local v redis.call(decr, KEYS[1]); if v 0 then redis.call(del, KEYS[1]) end; if v 0 then redis.call(publish, KEYS[2], ARGV[1]) end;,Arrays.ObjectasList(getRawName(), getChannelName()),CountDownLatchPubSub.ZERO_COUNT_MESSAGE);}...
}
countDownAsync()方法会执行decr命令将countDown的数量进行递减1。如果这个值已经小于等于0就执行del命令删除掉该CoutDownLatch。如果是这个值为0还会发布一条消息
publish redisson_countdownlatch__channel__{anyCountDownLatch} 0
