经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Zookeeper » 查看文章
zookeeper【5】分布式锁
来源：cnblogs　　作者：撸码识途　　时间：2018/11/28 9:58:24　　对本文有异议

我们常说的锁是单进程多线程锁，在多线程并发编程中，用于线程之间的数据同步，保护共享资源的访问。而分布式锁，指在分布式环境下，保护跨进程、跨主机、跨网络的共享资源，实现互斥访问，保证一致性。

 

架构图：

 

分布式锁获取思路
a、在获取分布式锁的时候在locker节点下创建临时顺序节点，释放锁的时候删除该临时节点。

b、客户端调用createNode方法在locker下创建临时顺序节点，然后调用getChildren(“locker”)来获取locker下面的所有子节点，注意此时不用设置任何Watcher。

c、客户端获取到所有的子节点path之后，如果发现自己创建的子节点序号最小，那么就认为该客户端获取到了锁。

d、如果发现自己创建的节点并非locker所有子节点中最小的，说明自己还没有获取到锁，此时客户端需要找到比自己小的那个节点，然后对其调用exist()方法，同时对其注册事件监听器。

e、之后，让这个被关注的节点删除，则客户端的Watcher会收到相应通知，此时再次判断自己创建的节点是否是locker子节点中序号最小的，如果是则获取到了锁，如果不是则重复以上步骤继续获取到比自己小的一个节点并注册监听。

 

实现代码：

  1. import org.I0Itec.zkclient.IZkDataListener;
  2. import org.I0Itec.zkclient.ZkClient;
  3. import org.I0Itec.zkclient.exception.ZkNoNodeException;
  5. import java.util.Collections;
  6. import java.util.Comparator;
  7. import java.util.List;
  8. import java.util.concurrent.CountDownLatch;
  9. import java.util.concurrent.TimeUnit;
  11. public class BaseDistributedLock {
  13.     private final ZkClientExt client;
  14.     private final String  path;
  15.     private final String  basePath;
  16.     private final String  lockName;
  17.     private static final Integer  MAX_RETRY_COUNT = 10;
  19.     public BaseDistributedLock(ZkClientExt client, String path, String lockName){
  21.         this.client = client;
  22.         this.basePath = path;
  23.         this.path = path.concat("/").concat(lockName);
  24.         this.lockName = lockName;
  26.     }
  28.     // 删除成功获取锁之后所创建的那个顺序节点
  29.     private void deleteOurPath(String ourPath) throws Exception{
  30.         client.delete(ourPath);
  31.     }
  33.     // 创建临时顺序节点
  34.     private String createLockNode(ZkClient client, String path) throws Exception{
  35.         return client.createEphemeralSequential(path, null);
  36.     }
  38.     // 等待比自己次小的顺序节点的删除
  39.     private boolean waitToLock(long startMillis, Long millisToWait, String ourPath) throws Exception{
  41.         boolean  haveTheLock = false;
  42.         boolean  doDelete = false;
  44.         try {
  46.             while ( !haveTheLock ) {
  47.                 // 获取/locker下的经过排序的子节点列表
  48.                 List<String> children = getSortedChildren();
  50.                 // 获取刚才自己创建的那个顺序节点名
  51.                 String sequenceNodeName = ourPath.substring(basePath.length()+1);
  53.                 // 判断自己排第几个
  54.                 int  ourIndex = children.indexOf(sequenceNodeName);
  55.                 if (ourIndex < 0){ // 网络抖动，获取到的子节点列表里可能已经没有自己了
  56.                     throw new ZkNoNodeException("节点没有找到: " + sequenceNodeName);
  57.                 }
  59.                 // 如果是第一个，代表自己已经获得了锁
  60.                 boolean isGetTheLock = ourIndex == 0;
  62.                 // 如果自己没有获得锁，则要watch比我们次小的那个节点
  63.                 String  pathToWatch = isGetTheLock ? null : children.get(ourIndex - 1);
  65.                 if ( isGetTheLock ){
  66.                     haveTheLock = true;
  68.                 } else {
  70.                     // 订阅比自己次小顺序节点的删除事件
  71.                     String  previousSequencePath = basePath .concat( "/" ) .concat( pathToWatch );
  72.                     final CountDownLatch latch = new CountDownLatch(1);
  73.                     final IZkDataListener previousListener = new IZkDataListener() {
  75.                         public void handleDataDeleted(String dataPath) throws Exception {
  76.                             latch.countDown(); // 删除后结束latch上的await
  77.                         }
  79.                         public void handleDataChange(String dataPath, Object data) throws Exception {
  80.                             // ignore
  81.                         }
  82.                     };
  84.                     try {
  85.                         //订阅次小顺序节点的删除事件，如果节点不存在会出现异常
  86.                         client.subscribeDataChanges(previousSequencePath, previousListener);
  88.                         if ( millisToWait != null ) {
  89.                             millisToWait -= (System.currentTimeMillis() - startMillis);
  90.                             startMillis = System.currentTimeMillis();
  91.                             if ( millisToWait <= 0 ) {
  92.                                 doDelete = true;    // timed out - delete our node
  93.                                 break;
  94.                             }
  96.                             latch.await(millisToWait, TimeUnit.MICROSECONDS); // 在latch上await
  97.                         } else {
  98.                             latch.await(); // 在latch上await
  99.                         }
  101.                         // 结束latch上的等待后，继续while重新来过判断自己是否第一个顺序节点
  102.                     }
  103.                     catch ( ZkNoNodeException e ) {
  104.                         //ignore
  105.                     } finally {
  106.                         client.unsubscribeDataChanges(previousSequencePath, previousListener);
  107.                     }
  109.                 }
  110.             }
  111.         }
  112.         catch ( Exception e ) {
  113.             //发生异常需要删除节点
  114.             doDelete = true;
  115.             throw e;
  116.         } finally {
  117.             //如果需要删除节点
  118.             if ( doDelete ) {
  119.                 deleteOurPath(ourPath);
  120.             }
  121.         }
  122.         return haveTheLock;
  123.     }
  125.     private String getLockNodeNumber(String str, String lockName) {
  126.         int index = str.lastIndexOf(lockName);
  127.         if ( index >= 0 ) {
  128.             index += lockName.length();
  129.             return index <= str.length() ? str.substring(index) : "";
  130.         }
  131.         return str;
  132.     }
  134.     // 获取/locker下的经过排序的子节点列表
  135.     List<String> getSortedChildren() throws Exception {
  136.         try{
  138.             List<String> children = client.getChildren(basePath);
  139.             Collections.sort(
  140.                     children, new Comparator<String>() {
  141.                         public int compare(String lhs, String rhs) {
  142.                             return getLockNodeNumber(lhs, lockName).compareTo(getLockNodeNumber(rhs, lockName));
  143.                         }
  144.                     }
  145.             );
  146.             return children;
  148.         } catch (ZkNoNodeException e){
  149.             client.createPersistent(basePath, true);
  150.             return getSortedChildren();
  151.         }
  152.     }
  154.     protected void releaseLock(String lockPath) throws Exception{
  155.         deleteOurPath(lockPath);
  156.     }
  158.     protected String attemptLock(long time, TimeUnit unit) throws Exception {
  160.         final long      startMillis = System.currentTimeMillis();
  161.         final Long      millisToWait = (unit != null) ? unit.toMillis(time) : null;
  163.         String          ourPath = null;
  164.         boolean         hasTheLock = false;
  165.         boolean         isDone = false;
  166.         int             retryCount = 0;
  168.         //网络闪断需要重试一试
  169.         while ( !isDone ) {
  170.             isDone = true;
  172.             try {
  173.                 // 在/locker下创建临时的顺序节点
  174.                 ourPath = createLockNode(client, path);
  175.                 // 判断自己是否获得了锁，如果没有获得那么等待直到获得锁或者超时
  176.                 hasTheLock = waitToLock(startMillis, millisToWait, ourPath);
  177.             } catch ( ZkNoNodeException e ) { // 捕获这个异常
  178.                 if ( retryCount++ < MAX_RETRY_COUNT ) { // 重试指定次数
  179.                     isDone = false;
  180.                 } else {
  181.                     throw e;
  182.                 }
  183.             }
  184.         }
  185.         if ( hasTheLock ) {
  186.             return ourPath;
  187.         }
  189.         return null;
  190.     }
  193. }
  1. import java.util.concurrent.TimeUnit;
  3. public interface DistributedLock {
  5.     /*
  6.      * 获取锁，如果没有得到就等待
  7.      */
  8.     public void acquire() throws Exception;
  10.     /*
  11.      * 获取锁，直到超时
  12.      */
  13.     public boolean acquire(long time, TimeUnit unit) throws Exception;
  15.     /*
  16.      * 释放锁
  17.      */
  18.     public void release() throws Exception;
  21. }
  1. import java.io.IOException;
  2. import java.util.concurrent.TimeUnit;
  4. public class SimpleDistributedLockMutex extends BaseDistributedLock implements
  5.         DistributedLock {
  7.     //锁名称前缀,成功创建的顺序节点如lock-0000000000,lock-0000000001,...
  8.     private static final String LOCK_NAME = "lock-";
  10.     // zookeeper中locker节点的路径
  11.     private final String basePath;
  13.     // 获取锁以后自己创建的那个顺序节点的路径
  14.     private String ourLockPath;
  16.     private boolean internalLock(long time, TimeUnit unit) throws Exception {
  18.         ourLockPath = attemptLock(time, unit);
  19.         return ourLockPath != null;
  21.     }
  23.     public SimpleDistributedLockMutex(ZkClientExt client, String basePath){
  25.         super(client,basePath,LOCK_NAME);
  26.         this.basePath = basePath;
  28.     }
  30.     // 获取锁
  31.     public void acquire() throws Exception {
  32.         if ( !internalLock(-1, null) ) {
  33.             throw new IOException("连接丢失!在路径:'"+basePath+"'下不能获取锁!");
  34.         }
  35.     }
  37.     // 获取锁，可以超时
  38.     public boolean acquire(long time, TimeUnit unit) throws Exception {
  40.         return internalLock(time, unit);
  41.     }
  43.     // 释放锁
  44.     public void release() throws Exception {
  46.         releaseLock(ourLockPath);
  47.     }
  50. }
  1. import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer;
  3. public class TestDistributedLock {
  5.     public static void main(String[] args) {
  7.         final ZkClientExt zkClientExt1 = new ZkClientExt("192.168.1.105:2181", 5000, 5000, new BytesPushThroughSerializer());
  8.         final SimpleDistributedLockMutex mutex1 = new SimpleDistributedLockMutex(zkClientExt1, "/Mutex");
  10.         final ZkClientExt zkClientExt2 = new ZkClientExt("192.168.1.105:2181", 5000, 5000, new BytesPushThroughSerializer());
  11.         final SimpleDistributedLockMutex mutex2 = new SimpleDistributedLockMutex(zkClientExt2, "/Mutex");
  13.         try {
  14.             mutex1.acquire();
  15.             System.out.println("Client1 locked");
  16.             Thread client2Thd = new Thread(new Runnable() {
  18.                 public void run() {
  19.                     try {
  20.                         mutex2.acquire();
  21.                         System.out.println("Client2 locked");
  22.                         mutex2.release();
  23.                         System.out.println("Client2 released lock");
  25.                     } catch (Exception e) {
  26.                         e.printStackTrace();
  27.                     }
  28.                 }
  29.             });
  30.             client2Thd.start();
  31.             Thread.sleep(5000);
  32.             mutex1.release();
  33.             System.out.println("Client1 released lock");
  35.             client2Thd.join();
  37.         } catch (Exception e) {
  39.             e.printStackTrace();
  40.         }
  42.     }
  44. }
  1. import org.I0Itec.zkclient.ZkClient;
  2. import org.I0Itec.zkclient.serialize.ZkSerializer;
  3. import org.apache.zookeeper.data.Stat;
  5. import java.util.concurrent.Callable;
  7. public class ZkClientExt extends ZkClient {
  9.     public ZkClientExt(String zkServers, int sessionTimeout, int connectionTimeout, ZkSerializer zkSerializer) {
  10.         super(zkServers, sessionTimeout, connectionTimeout, zkSerializer);
  11.     }
  13.     @Override
  14.     public void watchForData(final String path) {
  15.         retryUntilConnected(new Callable<Object>() {
  17.             public Object call() throws Exception {
  18.                 Stat stat = new Stat();
  19.                 _connection.readData(path, stat, true);
  20.                 return null;
  21.             }
  23.         });
  24.     }
  26. }

 

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Zookeeper热门文章

  • java Zookeeper简述
  • zookeeper-操作与应用场景-《每日五分钟搞定大数据》
  • ZooInspector 连接不到 Zookeeper 的解决方法
  • IDEA zookeeper插件的使用
  • 解决dubbo错误ip及ip乱入问题的方法
  • zookeeper集群搭建
  • 大数据开发phoenix连接hbase流程详解
  • 快来看看 Dubbo 是如何实现服务的动态发现的
  • zookeeper特性与节点说明
  • zookeeper【1】配置管理

    • Zookeeper推荐文章

  • Zookeeper安装部署
  • 得物 ZooKeeper SLA 也可以 99.99%
  • zookeeper源码(06)ZooKeeperServer及子类
  • Dubbo 我手写几行代码，就把通信模式给你解释清楚！
  • Rpc-实现Zookeeper注册中心
  • Dubbo 中 Zookeeper 注册中心原理分析
  • Java&nbsp;zookeeper服务的使用详解
  • 【入门】ZooKeeper 相关概念总结
  • zookeeper特性与节点说明
  • 快来看看 Dubbo 是如何实现服务的动态发现的

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号