前言

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

使用 Demo

  1. public class CuratorDemo {
  2. public static void main(String[] args) throws Exception {
  3. String connectString = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183";
  4. RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
  5. CuratorFramework client = CuratorFrameworkFactory
  6. .builder()
  7. .connectString(connectString)
  8. .retryPolicy(retryPolicy)
  9. .build();
  10. client.start();
  11. InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/semaphores/semaphore_01", 3);
  12. for (int i = 0; i < 10; i++) {
  13. new Thread(() -> {
  14. try {
  15. System.out.println(Thread.currentThread() + " 线程 start - " + LocalTime.now());
  16. Lease lease = semaphore.acquire();
  17. System.out.println(Thread.currentThread() + " 线程 execute - " + LocalTime.now());
  18. Thread.sleep(3000);
  19. System.out.println(Thread.currentThread() + " 线程 over -" + LocalTime.now());
  20. semaphore.returnLease(lease);
  21. } catch (Exception e) {
  22. }
  23. }).start();
  24. }
  25. Thread.sleep(1000000);
  26. }
  27. }

控制台输出数据如下:
4-Semaphore和互斥锁 - 图1

源码

获取凭证

核心源码:InterProcessSemaphoreV2#internalAcquire1Lease
这里仅介绍大概逻辑,有兴趣的小伙伴可以自行阅读源码。
4-Semaphore和互斥锁 - 图2
lock 是 InterProcessMutexInterProcessSemaphoreV2 信号量,也是借助于最基础的加锁。
4-Semaphore和互斥锁 - 图3
通过图也可以看出,使用 InterProcessSemaphoreV2 时,会先创建 /semaphores/semaphore_01 路径,并在路径下创建 locks 节点。也就是 /semaphores/semaphore_01/locks 路径下,有 10 个临时顺序节点。
4-Semaphore和互斥锁 - 图4
紧接着会在 /semaphores/semaphore_01 路径下创建 leases 节点,所以创建锁的临时顺序节点之后,会紧接着在 /semaphores/semaphore_01/leases 下创建临时顺序节点。
4-Semaphore和互斥锁 - 图5
/semaphores/semaphore_01/leases 节点进行监听,同时获取 /semaphores/semaphore_01/leases 下面的子节点数量。

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

    释放凭证

    4-Semaphore和互斥锁 - 图6
    释放凭证就是调用 Lease 的 close 方法,删除节点,这样 /semaphores/semaphore_01/leases 上的监听器就会触发,然后其他线程获取凭证。

    互斥锁

    互斥锁 InterProcessSemaphoreMutex,不支持重入,其他的和可重入锁并没有什么区别。就是基于 InterProcessSemaphoreV2 实现的。
    4-Semaphore和互斥锁 - 图7
    就是把计数的值 maxLeases 设置为了 1。

    总结

    信号量 InterProcessSemaphoreV2 其实是通过判断节点下的子节点数量来实现控制信号量,同时内部加锁是基于可重入锁 InterProcessMutex 实现的。
    互斥锁 InterProcessSemaphoreMutex 则是将信号量的技术设置为 1 来实现互斥功能。

    相关推荐