前言

分布式信号量，之前在 Redisson 中也介绍过，Redisson 的信号量是将计数维护在 Redis 中的，那现在来看一下 Curator 是如何基于 ZooKeeper 实现信号量的。

使用 Demo

public class CuratorDemo {

    public static void main(String[] args) throws Exception {

        String connectString = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183";

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

        CuratorFramework client = CuratorFrameworkFactory
                .builder()
                .connectString(connectString)
                .retryPolicy(retryPolicy)
                .build();
        client.start();

        InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/semaphores/semaphore_01", 3);

        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread() + " 线程 start - " + LocalTime.now());
                    Lease lease = semaphore.acquire();
                    System.out.println(Thread.currentThread() + " 线程 execute - " + LocalTime.now());
                    Thread.sleep(3000);
                    System.out.println(Thread.currentThread() + " 线程 over -" + LocalTime.now());
                    semaphore.returnLease(lease);
                } catch (Exception e) {

                }

            }).start();
        }

        Thread.sleep(1000000);

    }
}

控制台输出数据如下：

源码

获取凭证

核心源码：InterProcessSemaphoreV2#internalAcquire1Lease

这里仅介绍大概逻辑，有兴趣的小伙伴可以自行阅读源码。

lock 是 InterProcessMutex，InterProcessSemaphoreV2 信号量，也是借助于最基础的加锁。

通过图也可以看出，使用 InterProcessSemaphoreV2 时，会先创建 /semaphores/semaphore_01 路径，并在路径下创建 locks 节点。也就是 /semaphores/semaphore_01/locks 路径下，有 10 个临时顺序节点。

紧接着会在 /semaphores/semaphore_01 路径下创建 leases 节点，所以创建锁的临时顺序节点之后，会紧接着在 /semaphores/semaphore_01/leases 下创建临时顺序节点。

对 /semaphores/semaphore_01/leases 节点进行监听，同时获取 /semaphores/semaphore_01/leases 下面的子节点数量。

1. 如果子节点数量小于等于信号量计数，则直接结束循环；
2. 如果大于，则会进入 wait 等待唤醒。

释放凭证

释放凭证就是调用 Lease 的 close 方法，删除节点，这样 /semaphores/semaphore_01/leases 上的监听器就会触发，然后其他线程获取凭证。

互斥锁

互斥锁 InterProcessSemaphoreMutex，不支持重入，其他的和可重入锁并没有什么区别。就是基于 InterProcessSemaphoreV2 实现的。

就是把计数的值 maxLeases 设置为了 1。

总结

信号量 InterProcessSemaphoreV2 其实是通过判断节点下的子节点数量来实现控制信号量，同时内部加锁是基于可重入锁 InterProcessMutex 实现的。

互斥锁 InterProcessSemaphoreMutex 则是将信号量的技术设置为 1 来实现互斥功能。

相关推荐

• ZooKeeper 分布式锁 Curator 源码 03：可重入锁并发加锁
• ZooKeeper 分布式锁 Curator 源码 02：可重入锁重复加锁和锁释放
• ZooKeeper 分布式锁 Curator 源码 01：可重入锁