devbox
寸_铁
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Zookeeper开源客户端curator 分布式锁_locked <0x00000000fc66e180> (a org.apache.curator.

作者：寸_铁 | 2024-08-11 19:25:35

赞

踩

locked <0x00000000fc66e180> (a org.apache.curator.framework.recipes.leader.l

目录

一、概述

二、InterProcessMutex分布式可重入排它锁

三、InterProcessReadWriteLock读写锁

四、总结

一、概述

前面我们已经介绍过zookeeper是分布式锁的实现方案之一，它是利用临时有序节点的特性来保证唯一性的。

Zookeeper分布式锁原理：

  • 获取锁

在zookeeper中，一个分布式锁对应到zookeeper中的一个文件夹，每个客户端线程请求获取分布式锁的时候，就需要在这个文件夹创建一个临时有序节点，有两种情况：

  • 1）、创建的临时顺序节点是文件夹下的第一个节点，则认为是获取分布式锁成功；
  • 2）、创建的临时顺序节点不是文件夹下的第一个节点，则认为当前锁已经被另一个客户端线程获取，此时当前线程进入阻塞等待，这个时候它会监听上一个节点【序号小于当前节点的节点】，等待前一个节点释放锁的时候唤醒当前线程；

阻塞-唤醒逻辑：把文件夹下的节点顺序排一下序，找到当前节点的前一个节点，使用watcher机制监听前面一个节点的变化，当前一个节点被删除时会触发Watch事件，进而唤醒当前阻塞的线程。

如果前一个节点对应的客户端崩溃了，则节点对应的Watch事件也会触发，也会唤醒后一个节点对应的客户端线程，此时仍需要判断当前节点是第一个节点之后才能获取锁，否则继续进入阻塞并监听前面一个节点。

  • 可重入性

当某个客户端线程第一次获取锁成功之后，在JVM内存中的一个ConcurrentMap中存储当前线程对应的锁路径及重入次数，后面同一个线程再次获取锁时，先检查该Map中当前锁是否已被当前线程占用即可，如果已占用，则只需要递增重入次数即可。

注意：只考虑同一个客户端、同一个线程获取同一个分布式锁的可重入性。

  • 释放锁

释放锁时，首先将锁的重入次数减一，然后判断重入次数是否已经为0：

  • 1）如果重入次数为0，则删除当前客户端线程对应的临时顺序节点，删除操作会触发此节点的Watch事件，如果有别的客户端线程正在阻塞等待，则会通过Watch机制唤醒；
  • 2）如果重入次数非0，则说明还未完全释放锁，直接返回即可；

二、InterProcessMutex分布式可重入排它锁

  1. package com.wsh.zookeeper.zookeeperapidemo.curator;
  2.  
  3. import org.apache.curator.framework.CuratorFramework;
  4. import org.apache.curator.framework.CuratorFrameworkFactory;
  5. import org.apache.curator.framework.recipes.locks.InterProcessLock;
  6. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
  7. import org.apache.curator.retry.ExponentialBackoffRetry;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10.  
  11. public class CuratorInterProcessMutexDemo {
  12.  
  13.     private static final Logger logger = LoggerFactory.getLogger(CuratorInterProcessMutexDemo.class);
  14.     private static final String ZOOKEEPER_SERVER_ADDRESS = "192.168.179.128:2181,192.168.179.129:2181,192.168.179.133:2181";
  15.  
  16.     /**
  17.      * Curator对象
  18.      */
  19.     private static CuratorFramework curatorFramework;
  20.  
  21.     static {
  22.         //初始化curator对象
  23.         curatorFramework = CuratorFrameworkFactory
  24.                 .builder()
  25.                 //会话超时时间
  26.                 .sessionTimeoutMs(5000)
  27.                 //服务器集群地址
  28.                 .connectString(ZOOKEEPER_SERVER_ADDRESS)
  29.                 //重试策略
  30.                 .retryPolicy(new ExponentialBackoffRetry(1000, 3))
  31.                 .build();
  32.         //开启客户端
  33.         curatorFramework.start();
  34.     }
  35.  
  36.  
  37.     private static void getLock(InterProcessLock interProcessLock) {
  38.         try {
  39.             interProcessLock.acquire();
  40.             logger.info("线程：" + Thread.currentThread().getName() + "获取锁成功......");
  41.         } catch (Exception e) {
  42.             e.printStackTrace();
  43.         } finally {
  44.             try {
  45.                 interProcessLock.release();
  46.                 logger.info("线程：" + Thread.currentThread().getName() + "释放锁成功......");
  47.             } catch (Exception e) {
  48.                 e.printStackTrace();
  49.             }
  50.         }
  51.     }
  52.  
  53.     public static void main(String[] args) {
  54.         //分布式可重入排它锁
  55.         InterProcessMutex interProcessMutex = new InterProcessMutex(curatorFramework, "/distributeLock");
  56.         new Thread(() -> getLock(interProcessMutex), "T1").start();
  57.         new Thread(() -> getLock(interProcessMutex), "T2").start();
  58.         new Thread(() -> getLock(interProcessMutex), "T3").start();
  59.         new Thread(() -> getLock(interProcessMutex), "T4").start();
  60.     }
  61.  
  62. }

运行程序，观察后端日志：

  1. 14:41:37.532 [T2] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessMutexDemo - 线程：T2获取锁成功......
  2. 14:41:37.538 [T2] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessMutexDemo - 线程：T2释放锁成功......
  3.  
  4. 14:41:37.543 [T4] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessMutexDemo - 线程：T4获取锁成功......
  5. 14:41:37.554 [T4] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessMutexDemo - 线程：T4释放锁成功......
  6.  
  7. 14:41:37.555 [T3] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessMutexDemo - 线程：T3获取锁成功......
  8. 14:41:37.559 [T3] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessMutexDemo - 线程：T3释放锁成功......
  9.  
  10. 14:41:37.560 [T1] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessMutexDemo - 线程：T1获取锁成功......
  11. 14:41:37.564 [T1] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessMutexDemo - 线程：T1释放锁成功......

如上可见，多个线程争夺一个共享锁的时候，相互之间是互斥的。流程分析如下图所示：

三、InterProcessReadWriteLock读写锁

curator实现了跨JVM的可重入读写互斥锁。读写锁包含了读锁、写锁两个，它们的互斥关系如下：

  • 1）读写互斥；
  • 2）写写互斥；
  • 3）读读不互斥；

案例：读写互斥

  1. package com.wsh.zookeeper.zookeeperapidemo.curator;
  2.  
  3. import org.apache.curator.framework.CuratorFramework;
  4. import org.apache.curator.framework.CuratorFrameworkFactory;
  5. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
  6. import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
  7. import org.apache.curator.retry.ExponentialBackoffRetry;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10.  
  11. public class CuratorInterProcessReadWriteLockDemo {
  12.  
  13.     private static final Logger logger = LoggerFactory.getLogger(CuratorInterProcessReadWriteLockDemo.class);
  14.     private static final String ZOOKEEPER_SERVER_ADDRESS = "192.168.179.128:2181,192.168.179.129:2181,192.168.179.133:2181";
  15.  
  16.     /**
  17.      * Curator对象
  18.      */
  19.     private static CuratorFramework curatorFramework;
  20.     /**
  21.      * curator读锁对象
  22.      */
  23.     private static InterProcessMutex readLock;
  24.     /**
  25.      * curator写锁对象
  26.      */
  27.     private static InterProcessMutex writeLock;
  28.  
  29.     static {
  30.         //初始化curator对象
  31.         curatorFramework = CuratorFrameworkFactory
  32.                 .builder()
  33.                 //会话超时时间
  34.                 .sessionTimeoutMs(5000)
  35.                 //服务器集群地址
  36.                 .connectString(ZOOKEEPER_SERVER_ADDRESS)
  37.                 //重试策略
  38.                 .retryPolicy(new ExponentialBackoffRetry(1000, 3))
  39.                 .build();
  40.         //开启客户端
  41.         curatorFramework.start();
  42.         InterProcessReadWriteLock interProcessReadWriteLock = new InterProcessReadWriteLock(curatorFramework, "/distribute-lock");
  43.         readLock = interProcessReadWriteLock.readLock();
  44.         writeLock = interProcessReadWriteLock.writeLock();
  45.     }
  46.  
  47.     public static void main(String[] args) {
  48.         new Thread(() -> {
  49.             try {
  50.                 readLock.acquire();
  51.                 logger.info("线程：" + Thread.currentThread().getName() + "获取读锁...");
  52.                 Thread.sleep(3000);
  53.             } catch (Exception e) {
  54.                 e.printStackTrace();
  55.             } finally {
  56.                 try {
  57.                     readLock.release();
  58.                     logger.info("线程：" + Thread.currentThread().getName() + "释放读锁...");
  59.                 } catch (Exception e) {
  60.                     e.printStackTrace();
  61.                 }
  62.             }
  63.         }, "T1").start();
  64.  
  65.         new Thread(() -> {
  66.             try {
  67.                 Thread.sleep(2000);
  68.                 writeLock.acquire();
  69.                 logger.info("线程：" + Thread.currentThread().getName() + "获取写锁...");
  70.             } catch (Exception e) {
  71.                 e.printStackTrace();
  72.             } finally {
  73.                 try {
  74.                     writeLock.release();
  75.                     logger.info("线程：" + Thread.currentThread().getName() + "释放写锁...");
  76.                 } catch (Exception e) {
  77.                     e.printStackTrace();
  78.                 }
  79.             }
  80.         }, "T2").start();
  81.     }
  82.  
  83. }
  1. 15:22:11.460 [T1] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T1获取读锁...
  2. 15:22:14.468 [T1] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T1释放读锁...
  3.  
  4. 15:22:14.470 [T2] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T2获取写锁...
  5. 15:22:14.473 [T2] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T2释放写锁...

由上日志可见，读锁先获取到锁资源后，写锁必须等待锁释放之后，才能获取锁。

案例：读读不互斥

  1. package com.wsh.zookeeper.zookeeperapidemo.curator;
  2.  
  3. import org.apache.curator.framework.CuratorFramework;
  4. import org.apache.curator.framework.CuratorFrameworkFactory;
  5. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
  6. import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
  7. import org.apache.curator.retry.ExponentialBackoffRetry;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10.  
  11. public class CuratorInterProcessReadWriteLockDemo {
  12.  
  13.     private static final Logger logger = LoggerFactory.getLogger(CuratorInterProcessReadWriteLockDemo.class);
  14.     private static final String ZOOKEEPER_SERVER_ADDRESS = "192.168.179.128:2181,192.168.179.129:2181,192.168.179.133:2181";
  15.  
  16.     /**
  17.      * Curator对象
  18.      */
  19.     private static CuratorFramework curatorFramework;
  20.     /**
  21.      * curator读锁对象
  22.      */
  23.     private static InterProcessMutex readLock;
  24.     /**
  25.      * curator写锁对象
  26.      */
  27.     private static InterProcessMutex writeLock;
  28.  
  29.     static {
  30.         //初始化curator对象
  31.         curatorFramework = CuratorFrameworkFactory
  32.                 .builder()
  33.                 //会话超时时间
  34.                 .sessionTimeoutMs(5000)
  35.                 //服务器集群地址
  36.                 .connectString(ZOOKEEPER_SERVER_ADDRESS)
  37.                 //重试策略
  38.                 .retryPolicy(new ExponentialBackoffRetry(1000, 3))
  39.                 .build();
  40.         //开启客户端
  41.         curatorFramework.start();
  42.         InterProcessReadWriteLock interProcessReadWriteLock = new InterProcessReadWriteLock(curatorFramework, "/distribute-lock");
  43.         readLock = interProcessReadWriteLock.readLock();
  44.         writeLock = interProcessReadWriteLock.writeLock();
  45.     }
  46.  
  47.     public static void main(String[] args) {
  48.         new Thread(() -> {
  49.             try {
  50.                 readLock.acquire();
  51.                 logger.info("线程：" + Thread.currentThread().getName() + "获取读锁...");
  52.             } catch (Exception e) {
  53.                 e.printStackTrace();
  54.             } finally {
  55.                 try {
  56.                     readLock.release();
  57.                     logger.info("线程：" + Thread.currentThread().getName() + "释放读锁...");
  58.                 } catch (Exception e) {
  59.                     e.printStackTrace();
  60.                 }
  61.             }
  62.         }, "T1").start();
  63.  
  64.         new Thread(() -> {
  65.             try {
  66.                 readLock.acquire();
  67.                 logger.info("线程：" + Thread.currentThread().getName() + "获取读锁...");
  68.             } catch (Exception e) {
  69.                 e.printStackTrace();
  70.             } finally {
  71.                 try {
  72.                     readLock.release();
  73.                     logger.info("线程：" + Thread.currentThread().getName() + "释放读锁...");
  74.                 } catch (Exception e) {
  75.                     e.printStackTrace();
  76.                 }
  77.             }
  78.         }, "T2").start();
  79.     }
  80.  
  81. }
  1. 15:32:30.347 [T2] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T2获取读锁...
  2. 15:32:30.347 [T1] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T1获取读锁...
  3.  
  4. 15:32:30.354 [T1] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T1释放读锁...
  5. 15:32:30.354 [T2] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T2释放读锁...

 由上面的日志可见，两个读线程之间获取锁资源的话，是可以同时获取到锁的，并不互斥。

案例：写写互斥

  1. package com.wsh.zookeeper.zookeeperapidemo.curator;
  2.  
  3. import org.apache.curator.framework.CuratorFramework;
  4. import org.apache.curator.framework.CuratorFrameworkFactory;
  5. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
  6. import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
  7. import org.apache.curator.retry.ExponentialBackoffRetry;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10.  
  11. public class CuratorInterProcessReadWriteLockDemo {
  12.  
  13.     private static final Logger logger = LoggerFactory.getLogger(CuratorInterProcessReadWriteLockDemo.class);
  14.     private static final String ZOOKEEPER_SERVER_ADDRESS = "192.168.179.128:2181,192.168.179.129:2181,192.168.179.133:2181";
  15.  
  16.     /**
  17.      * Curator对象
  18.      */
  19.     private static CuratorFramework curatorFramework;
  20.     /**
  21.      * curator读锁对象
  22.      */
  23.     private static InterProcessMutex readLock;
  24.     /**
  25.      * curator写锁对象
  26.      */
  27.     private static InterProcessMutex writeLock;
  28.  
  29.     static {
  30.         //初始化curator对象
  31.         curatorFramework = CuratorFrameworkFactory
  32.                 .builder()
  33.                 //会话超时时间
  34.                 .sessionTimeoutMs(5000)
  35.                 //服务器集群地址
  36.                 .connectString(ZOOKEEPER_SERVER_ADDRESS)
  37.                 //重试策略
  38.                 .retryPolicy(new ExponentialBackoffRetry(1000, 3))
  39.                 .build();
  40.         //开启客户端
  41.         curatorFramework.start();
  42.         InterProcessReadWriteLock interProcessReadWriteLock = new InterProcessReadWriteLock(curatorFramework, "/distribute-lock");
  43.         readLock = interProcessReadWriteLock.readLock();
  44.         writeLock = interProcessReadWriteLock.writeLock();
  45.     }
  46.  
  47.     public static void main(String[] args) {
  48.         new Thread(() -> {
  49.             try {
  50.                 writeLock.acquire();
  51.                 logger.info("线程：" + Thread.currentThread().getName() + "获取写锁...");
  52.             } catch (Exception e) {
  53.                 e.printStackTrace();
  54.             } finally {
  55.                 try {
  56.                     writeLock.release();
  57.                     logger.info("线程：" + Thread.currentThread().getName() + "释放写锁...");
  58.                 } catch (Exception e) {
  59.                     e.printStackTrace();
  60.                 }
  61.             }
  62.         }, "T1").start();
  63.  
  64.         new Thread(() -> {
  65.             try {
  66.                 writeLock.acquire();
  67.                 logger.info("线程：" + Thread.currentThread().getName() + "获取写锁...");
  68.             } catch (Exception e) {
  69.                 e.printStackTrace();
  70.             } finally {
  71.                 try {
  72.                     writeLock.release();
  73.                     logger.info("线程：" + Thread.currentThread().getName() + "释放写锁...");
  74.                 } catch (Exception e) {
  75.                     e.printStackTrace();
  76.                 }
  77.             }
  78.         }, "T2").start();
  79.     }
  80.  
  81. }
  1. 15:33:28.701 [T1] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T1获取写锁...
  2. 15:33:28.706 [T1] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T1释放写锁...
  3.  
  4. 15:33:28.711 [T2] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T2获取写锁...
  5. 15:33:28.717 [T2] INFO com.wsh.zookeeper.zookeeperapidemo.curator.CuratorInterProcessReadWriteLockDemo - 线程：T2释放写锁...

如上日志可见，两个线程同时尝试获取锁资源的时候，某个时刻只会有一个线程能获取成功，其他线程必须等待释放锁后才能尝试去获取锁。

四、总结

本文主要总结了curator分布式锁的实现原理，以及通过案例详解介绍了curator的可重入排它锁以及读写锁的使用，在分布式系统开发中，实现分布式锁的方式有很多种，如数据库方式、redis、zookeeper等，具体根据业务场景选择合适的即可。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/寸_铁/article/detail/965714
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号