我们都知道JDK 并发包里面有CountDownLatch 类,这个类是干嘛用的?
简单的理解就是,我们主线程设置一个条件,然后阻塞,等其他线程去将条件达成,就可以让主线程继续执行下去
当然这里的条件就是数字
那么Redisson 是如何实现的?
public static void main(String[] args) throws Exception {
//构建一个配置信息对象
Config config = new Config();
config.useClusterServers()
//定时扫描连接信息 默认1000ms
.setScanInterval(2000)
.addNodeAddress("redis://127.0.0.1:7001");
//因为Redisson 是基于redis封装的一套便于复杂操作的框架
//所以这里构建对象肯定是创建一些与redis的连接
RedissonClient redisson = Redisson.create(config);
final RCountDownLatch countDownLatch = redisson.getCountDownLatch("countDownLatch");
countDownLatch.trySetCount(3);
//启动异步线程 执行countDown
for (int i = 0; i < 3; i++) {
new Thread(new Runnable() {
public void run() {
countDownLatch.countDown();
}
}).start();
}
countDownLatch.await();
}
trySetCount(3) 设置一个count 为3
@Override
public RFuture<Boolean> trySetCountAsync(long count) {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
//去判定KEY 是否存在
//然后设置一下key count
//同时发布一个消息
"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.<Object>asList(getName(), getChannelName()), newCountMessage, count);
}
await方法,其实就是死循环 不断的从redis获取count的值, 知道count为0,才会继续让主线程往下执行
public void await() throws InterruptedException {
RFuture<RedissonCountDownLatchEntry> future = subscribe();
try {
commandExecutor.syncSubscription(future);
while (getCount() > 0) {
// waiting for open state
RedissonCountDownLatchEntry entry = getEntry();
if (entry != null) {
entry.getLatch().await();
}
}
} finally {
unsubscribe(future);
}
}
countDown()方法
@Override
public RFuture<Void> countDownAsync() {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
//就是将对应的key 的值-1
"local v = redis.call('decr', KEYS[1]);" +
"if v <= 0 then redis.call('del', KEYS[1]) end;" +
//如果为0了,那么发布一跳信息到channel中去
"if v == 0 then redis.call('publish', KEYS[2], ARGV[1]) end;",
Arrays.<Object>asList(getName(), getChannelName()), zeroCountMessage);
}
其实看了源码,这个countDownLatch 其实非常的简单和清晰
其实最主要的还是redis这个高并发的共享存储,为什么这么说呢?以前jdk中的countDownLatch的可见性只有同一个进程中的线程,对吧,但是我们现在一般是分布式,跨机器,跨服务,那我们怎么去实现?
其实就是让对应的信息能让每个相关的服务能读取 操作 和实现对应的功能就行了,不管是用redis 还是其他可共享的存储服务,都可以,只是刚好redis 本身的性质比较符合这块的功能实现。