Redisson 源码初探（十一） CountDownLatch

浅浅的花香味﹌ 2022-11-03 11:13 39阅读 0赞

我们都知道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 本身的性质比较符合这块的功能实现。