Redisson 源码初探（十一） CountDownLatch

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