• Zookeeper系列——2Zookeeper应用及常用命令

    学习目标

    1. 了解Zookeeper原生核心API的使用

    2. 掌握Curator的使用

    第1章 常用命令

    zk的应用主要是针对三类：

    • java原生zk客户端的API操作（不用去学这部分内容，会增加太多的学习成本，了解一下就好了）

    • zkClient的使用，它是对Zookeeper原生API的封装

    • Apache Curator，也是对Zookeeper API 的封装（本文讲的应用针对这部分内容）

    在学Java API之前，我们先来了解一下zookeeper的常用命令

    1、连接zookeeper server

    [root@jt2 bin]# sh zkCli.sh -server 127.0.0.1:2181

    2、获取帮助help

     3、连接远程节点

    connect 192.168.8.75:2181

    4、关闭连接

    close

     5、显示集群

    1. [zk: localhost:2181(CONNECTED) 0] config
    2. server.0=jt2:2888:3888:participant
    3. server.1=jt3:2888:3888:participant
    4. server.2=jt4:2888:3888:participant
    5. version=0

    6、创建一个znode

    1. 命令语法：create [-s] [-e] [-c] [-t ttl] path [data] [acl]
    2.  
    3. -s：创建的是带序列号的节点，序列号用0填充节点路径。
    4. -e：创建的是临时节点。
    5. -c：创建的是容器节点
    6. path：znode的路径，ZooKeeper中没有相对路径，所有路径都必须以’/'开头。
    7. data：znode携带的数据。
    8. acl：这个节点的ACL。
    10. #创建一个永久节点
    11. [zk: localhost:2181(CONNECTED) 2] create /zkBase
    12. Created /zkBase
    13. #创建一个临时节点
    14. [zk: localhost:2181(CONNECTED) 3] create -e /ephemeral_node
    15. Created /ephemeral_node

    7、删除znode节点

    1. #删除节点前要求节点目录为空，不存在子节点
    2. [zk: localhost:2181(CONNECTED) 34] delete /config
    3. Node not empty: /config
    4. [zk: localhost:2181(CONNECTED) 35] delete /config/topics/test
    5. [zk: localhost:2181(CONNECTED) 27] delete /ephemeral_node
    6. #如果要删除整个节点及子节点可以使用deleteall
    7. [zk: 192.168.0.143:2181(CONNECTED) 36] deleteall /config

    8、显示一个节点的状态

    1. [zk: localhost:2181(CONNECTED) 11] stat /test
    2. cZxid = 0x180000000e
    3. ctime = Thu Jul 28 03:25:08 CST 2022
    4. mZxid = 0x180000000e
    5. mtime = Thu Jul 28 03:25:08 CST 2022
    6. pZxid = 0x180000000e
    7. cversion = 0
    8. dataVersion = 0
    9. aclVersion = 0
    10. ephemeralOwner = 0x0
    11. dataLength = 0
    12. numChildren = 0

    9、查看路径子节点

    命令语法：ls [-s] [-w] [-R] path

    • -s 同时显示stat信息

    • -w 只显示子节点信息，默认选项

    • -R 递归显示

    10、获取指定路径下的数据

    1. [zk: localhost:2181(CONNECTED) 16] get /zookeeper/config
    2. server.0=jt2:2888:3888:participant
    3. server.1=jt3:2888:3888:participant
    4. server.2=jt4:2888:3888:participant
    5. version=0

    11、设置或者更新路径数据

    1. [zk: localhost:2181(CONNECTED) 19] set /test/hehe "haha" 
    2. [zk: localhost:2181(CONNECTED) 20] get /test/hehe
    3. haha

    12、设置ACL

    ACL权限ACL 简写允许的操作
    CREATEc创建子节点
    READr获取节点的数据和它的子节点
    WRITEw设置节点的数据
    DELETEd删除子节点 （仅下一级节点）
    ADMINa设置 ACL 权限

    ZooKeeper内置了一些权限控制方案，可以用以下方案为每个节点设置权限：  

    方案描述
    world只有一个用户：anyone，代表所有人（默认）
    ip使用IP地址认证
    auth使用已添加认证的用户认证
    digest使用“用户名:密码”方式认证
    1. [zk: localhost:2181(CONNECTED) 21] getAcl /test
    2. 'world,'anyone
    3. : cdrwa
    4. [zk: localhost:2181(CONNECTED) 22] create /mynode1 hello
    5. Created /mynode1
    6. [zk: localhost:2181(CONNECTED) 23] addauth digest admin:admin
    7. [zk: localhost:2181(CONNECTED) 24] setAcl /mynode1 auth:admin:cdrwa
    8. [zk: localhost:2181(CONNECTED) 25] getAcl /mynode1
    9. 'digest,'admin:x1nq8J5GOJVPY6zgzhtTtA9izLc=
    10. : cdrwa

     13、同步数据集群间数据

    1. [zk: localhost:2181(CONNECTED) 26] sync /
    2. Sync is OK

    14、查看命令执行历史

    1. [zk: localhost:2181(CONNECTED) 27] history
    2. 17 - help
    3. 18 - getAllChildrenNumber /zookeeper
    4. 19 - set /test/hehe "haha"
    5. 20 - get /test/hehe
    6. 21 - getAcl /test
    7. 22 - create /mynode1 hello
    8. 23 - addauth digest admin:admin
    9. 24 - setAcl /mynode1 auth:admin:cdrwa
    10. 25 - getAcl /mynode1
    11. 26 - sync /
    12. 27 - history

    15、退出客户端

    1. [zk: localhost:2181(CONNECTED) 28] quit
    2.  
    3. WATCHER::
    4.  
    5. WatchedEvent state:Closed type:None path:null
    6. 2022-07-28 03:33:49,307 [myid:] - INFO  [main:ZooKeeper@1422] - Session: 0xcebb0001 closed
    7. 2022-07-28 03:33:49,308 [myid:] - INFO  [main-EventThread:ClientCnxn$EventThread@524] - EventThread shut down for session: 0xcebb0001

    第2章 Java API使用

    zookeeper客户端和服务器会话的建立是一个异步的过程，也就是说在程序中，程序方法在处理完客户端初始化后立即返回（即程序继续往下执行代码，这样，在大多数情况下并没有真正的构建好一个可用会话，在会话的生命周期处于“CONNECTED”时才算真正的建立完毕，所以需要使用到多线程中的一个工具类CountDownLatch）。

    2.1 创建会话

    （一共有4个构造方法，根据参数不同）

    1. Zookeeper(String connectString,int sessionTimeout,Watcher watcher)
    2. Zookeeper(String connectString,int sessionTimeout,Watcher watcher,boolean canBeReadOnly)
    3. Zookeeper(String connectString,int sessionTimeout,Watcher watcher,long sessionId,byte[] sessionPasswd)
    4. Zookeeper(String connectString,int sessionTimeout,Watcher watcher,long sessionId,byte[] sessionPasswd,boolean canBeReadOnly)

    参数说明：

    • connectString ：host:port指定的服务器列表，多个host:port之间用英文逗号分隔。还可以可选择的指定一个基路径，如果指定了一个基路径，则所有后续操作基于这个及路径进行。

    • sessionTimeOut：会话超时时间。以毫秒为单位。客户端和服务器端之间的连接通过心跳包进行维系，如果心跳包超过这个指定时间则认为会话超时失效。

    • watcher：指定默认观察者。如果为null表示不需要观察者。

    • canBeReadOnly ：是否支持只读服务。只当一个服务器失去过半连接后不能再进行写入操作时，是否继续支持读取操作。

    • sessionId、SessionPassword：会话编号 会话密码（通过两个确定唯一一台客户端），用来实现会话恢复（重复回话）。

    注意，整个创建会话的过程是异步的，构造方法会在初始化连接后即返回，并不代表真正建立好了一个会话，此时会话处于"CONNECTING"状态。当会话真正创建起来后，服务器会发送事件通知给客户端，只有客户端获取到这个通知后，会话才真正建立。

    代码演示

    1. public class ZkConnect implements Watcher {
    2.     private static final Logger log = LoggerFactory.getLogger(ZkConnect.class);
    3.     //public static final String zkServerPath = "192.168.8.74:2181,192.168.8.75:2181,192.168.8.76:2181";
    4.     public static final String zkServerPath = "127.0.0.1:2181";
    5.     public static final Integer timeout = 5000;
    6.     public static CountDownLatch countDownLatch = new CountDownLatch(1);
    7.  
    8.     /**
    9.      * 客户端与zkServer连接是一个异步的过程，当连接成功后，客户端会收到一个watch通知
    10.      * 参数：
    11.      *     connectString: 连接服务器的ip字符串
    12.      *     sessionTimeout: 超时时间，心跳收不到了，就超时
    13.      *     watcher: 通知事件，如果有对应的事件触发，则会收到一个通知；如果不需要，就设置为null
    14.      *     canBeReadOnly: 可读，当这个物理机节点断开后，还是可以读到数据的，只是不能写；此时数据被读取到的可能
    15.      *     是旧数据，此处建议设置为false
    16.      *     sessionId: 会话id
    17.      *     sessionPasswd: 会话密码，当会话丢失后，可以依据sessionId和sessionPasswd重新获取会话
    18.      */
    19.     public static void main(String[] args) throws Exception {
    20.         ZooKeeper zk = new ZooKeeper(zkServerPath, timeout, new ZkConnect());
    21.         log.warn("客户端开始连接zookeeper服务器。。。连接状态: {}", zk.getState());
    22.         countDownLatch.await(); // 如果不停顿一段时间, 会收不到watch通知
    23.         log.warn("连接状态: {}", zk.getState());
    24.     }
    25.  
    26.     @Override
    27.     public void process(WatchedEvent event) {
    28.         log.warn("接收到watch通知： {}", event);
    29.         countDownLatch.countDown();
    30.     }
    31. }
    1. public class ZkReconnect implements Watcher {
    2.     private static final Logger log = LogManager.getLogger(ZkReconnect.class);
    3.     public static final String zkServerPath = "127.0.0.1:2181";
    4.     public static final Integer timeout = 5000;
    5.     public static CountDownLatch countDownLatch1 = new CountDownLatch(1);
    6.     public static CountDownLatch countDownLatch2 = new CountDownLatch(2);
    7.  
    8.     public static void main(String[] args) throws Exception {
    9.         ZooKeeper zk = new ZooKeeper(zkServerPath, timeout, new ZkReconnect());
    10.         long sessionId = zk.getSessionId();
    11.         byte[] sessionPasswd = zk.getSessionPasswd();
    12.  
    13.         log.warn("客户端开始连接zookeeper服务器。。。连接状态: {}", zk.getState());
    14.         countDownLatch1.await(); // 如果不停顿一段时间, 会收不到watch通知
    15.         log.warn("连接状态: {}", zk.getState());
    16.  
    17.         Thread.sleep(1000);
    18.         log.warn("开始会话重连...");
    19.         ZooKeeper zkSession = new ZooKeeper(zkServerPath, timeout, new ZkReconnect(), sessionId, sessionPasswd);
    20.         log.warn("重新连接, 状态: {}", zk.getState());
    21.         countDownLatch2.await();
    22.         log.warn("重新连接, 状态: {}", zk.getState());
    23.     }
    24.  
    25.     @Override
    26.     public void process(WatchedEvent event) {
    27.         log.warn("接收到watch通知： {}", event);
    28.         countDownLatch1.countDown();
    29.         countDownLatch2.countDown();
    30.     }
    31. }

    2.2 创建节点

    提供了两套创建节点的方法，同步和异步创建节点方式。

    1. String create(final String path,byte data[],List acl,CreateMode createMode);//同步方式创建
    2. void create(final String path,byte data[],List acl,CreateMode createMode,StringCallback cb,Object ctx);//异步方式创建

    同步方式：

    path：节点路径（名称）：/nodeName。不允许递归创建节点，在父节点不存在的情况下，不允许创建子节点。

    data[]：节点内容：要求类型是字节数组，也就是说不支持序列话方式，如果需要实现序列化，可使用java相关序列化框架，如Hessian，Kryo。

    acl：节点权限：使用Ids.OPEN_ACL_UNSAFE开放权限即可。

    createMode:节点类型：创建节点的类型，CreateMode.*，提供了如下所示的四种节点类型：

    • PERSISTENT（持久节点）

    • PERSISTENT_SEQUENTIAL（持久顺序节点）

    • EPHEMERAL（临时节点，本次会话有效）

    • EPHEMERAL_SEQUENTIAL（临时顺序节点，本次会话有效）

    异步方式（在同步方法参数的基础上增加两个参数）：

    cb：回调方法：注册一个异步回调方法，要实现AsynCallBack.StringCallBack接口，重写processResult(int rc, String path, Object ctx, String name)方法，当节点创建完成后执行此方法。

    • rc：服务端响应码，0表示调用成功、-4表示端口连接、-110表示指定节点存在、-112表示会话已过期。

    • path：接口调用时传入的数据节点的路径参数。

    • ctx：调用接口传入的ctx值。

    • name：实际在服务端创建的节点的名称。

    ctx：传递给回调方法的参数，一般为上下文（Context）信息。 

    代码演示  

    1. public class ZkNodeCreate implements Watcher {
    2.     private ZooKeeper zooKeeper = null;
    3.     private static final Logger log = LoggerFactory.getLogger(ZkNodeCreate.class);
    4.     private static final String zkServerPath = "127.0.0.1:2181";
    5.     private static final Integer timeout = 5000;
    6.  
    7.     public ZkNodeCreate() {}
    8.     public ZkNodeCreate(String connectString) {
    9.         try {
    10.             zooKeeper = new ZooKeeper(connectString, timeout, new ZkNodeCreate());
    11.         } catch (Exception e) {
    12.             e.printStackTrace();
    13.             if (zooKeeper != null) {
    14.                 try {
    15.                     zooKeeper.close();
    16.                 } catch (Exception e1) {
    17.                     e1.printStackTrace();
    18.                 }
    19.             }
    20.         }
    21.     }
    22.  
    23.     public static void main(String[] args) throws InterruptedException {
    24.         ZkNodeCreate zkNodeOperator = new ZkNodeCreate(zkServerPath);
    25.         zkNodeOperator.createZKNode("/testnode", "testnode".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE);
    26.         new CountDownLatch(1).await();
    27.     }
    28.  
    29.     /**
    30.      * 同步或异步创建节点，都不支持子节点的递归创建，异步有一个callback函数
    31.      * 参数：
    32.      *     path: 创建的路径
    33.      *     data: 存储的数据
    34.      *     acl: 控制权限策略. Ids.OPEN_ACL_UNSAFE --> world:anyone:cdrwa
    35.      *                       Ids.CREATOR_ALL_ACL --> auth:user:password:cdrwa
    36.      *     createMode: 节点类型，是一个枚举
    37.      *                  PERSISTENT  持久节点
    38.      *                  PERSISTENT_SEQUENTIAL  持久顺序节点
    39.      *                  EPHEMERAL  临时节点
    40.      *                  EPHEMERAL_SEQUENTIAL  临时顺序节点
    41.      *
    42.      * @param path
    43.      * @param data
    44.      * @param acls
    45.      */
    46.     private void createZKNode(String path, byte[] data, ArrayList acls) {
    47.         String result = "";
    48.         try {
    49.             // 同步创建
    50.             //result = zooKeeper.create(path, data, acls, CreateMode.EPHEMERAL);
    51.             //log.warn("同步创建临时节点: {} 成功。。。", result);
    52.  
    53.             // 异步创建
    54.             String ctx = "{'create':'success'}";
    55.             zooKeeper.create(path, data, acls, CreateMode.EPHEMERAL, new CreateNodeCallBack(), ctx);
    56.             Thread.sleep(5000);
    57.             log.warn("异步创建临时节点: {} 成功。。。", result);
    58.  
    59.         } catch (Exception e) {
    60.             e.printStackTrace();
    61.         }
    62.     }
    63.  
    64.     @Override
    65.     public void process(WatchedEvent event) {
    66.         log.warn("客户端连接接收到watch通知： {}", event);
    67.     }
    68.  
    69.     public ZooKeeper getZooKeeper() {
    70.         return zooKeeper;
    71.     }
    72.  
    73.     public void setZooKeeper(ZooKeeper zooKeeper) {
    74.         this.zooKeeper = zooKeeper;
    75.     }
    76.  
    77.     private static class CreateNodeCallBack implements AsyncCallback.StringCallback {
    78.         @Override
    79.         public void processResult(int rc, String path, Object ctx, String name) {
    80.             log.warn("异步创建节点：{}, ctx: {}", path, (String)ctx);
    81.         }
    82.     }
    83. }

    2.3 节点操作

    1、修改节点数据

    1. public class ZkNodeUpdate implements Watcher {
    2.     private ZooKeeper zooKeeper = null;
    3.     private static final Logger log = LoggerFactory.getLogger(ZkNodeUpdate.class);
    4.     private static final String zkServerPath = "127.0.0.1:2181";
    5.     private static final Integer timeout = 5000;
    6.  
    7.     public ZkNodeUpdate() {}
    8.     public ZkNodeUpdate(String connectString) {
    9.         try {
    10.             zooKeeper = new ZooKeeper(connectString, timeout, new ZkNodeUpdate());
    11.         } catch (Exception e) {
    12.             e.printStackTrace();
    13.             if (zooKeeper != null) {
    14.                 try {
    15.                     zooKeeper.close();
    16.                 } catch (Exception e1) {
    17.                     e1.printStackTrace();
    18.                 }
    19.             }
    20.         }
    21.     }
    22.  
    23.     public static void main(String[] args) throws KeeperException, InterruptedException {
    24.         ZkNodeUpdate zkNodeOperator = new ZkNodeUpdate(zkServerPath);
    25.         // 创建节点
    26.         zkNodeOperator.createZKNode("/testnode", "testnode".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE);
    27.         // 修改节点数据  第三个参数是版本号dataVersion，用于乐观锁控制
    28.         Stat stat = zkNodeOperator.getZooKeeper().setData("/testnode", "修改后的数据".getBytes(), 0);
    29.         //zk.setData(path, data, version,new UpdateCallBack(),ctx);//异步修改
    30.         Thread.sleep(5000);
    31.         log.warn("修改后, dataVersion版本: {}", stat.getVersion());
    32.         new CountDownLatch(1).await();
    33.     }
    34.  
    35.     private void createZKNode(String path, byte[] data, ArrayList acls) {
    36.         String result = "";
    37.         try {
    38.             // 异步创建
    39.             String ctx = "{'create':'success'}";
    40.             zooKeeper.create(path, data, acls, CreateMode.EPHEMERAL, new CreateNodeCallBack(), ctx);
    41.             Thread.sleep(5000);
    42.             log.warn("异步创建临时节点: {} 成功。。。", result);
    43.         } catch (Exception e) {
    44.             e.printStackTrace();
    45.         }
    46.     }
    47.  
    48.     @Override
    49.     public void process(WatchedEvent event) {
    50.         log.warn("客户端连接接收到watch通知： {}", event);
    51.     }
    52.  
    53.     public ZooKeeper getZooKeeper() {
    54.         return zooKeeper;
    55.     }
    56.  
    57.     public void setZooKeeper(ZooKeeper zooKeeper) {
    58.         this.zooKeeper = zooKeeper;
    59.     }
    60.  
    61.     private static class CreateNodeCallBack implements AsyncCallback.StringCallback {
    62.         @Override
    63.         public void processResult(int rc, String path, Object ctx, String name) {
    64.             log.warn("异步创建节点：{}, ctx: {}", path, (String)ctx);
    65.         }
    66.     }
    67. }

    2、同步或异步删除节点数据

    1. public static void main(String[] args) throws KeeperException, InterruptedException {
    2.     ZkNodeDelete zkNodeOperator = new ZkNodeDelete(zkServerPath);
    3.     // 创建节点
    4.     zkNodeOperator.createZKNode("/testnode", "testnode".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE);
    5.  
    6.     // 同步删除节点
    7.     //zkNodeOperator.getZooKeeper().delete("/testnode", 1); // 第二个参数 dataVersion
    8.     Thread.sleep(5000);
    9.     // 异步删除节点
    10.     String ctx = "{'delete':'success'}";
    11.     zkNodeOperator.getZooKeeper().delete("/testnode", 0, new AsyncCallback.VoidCallback() {
    12.         @Override
    13.         public void processResult(int rc, String path, Object ctx) {
    14.             log.warn("异步删除节点：{}, ctx: {}", path, (String)ctx);
    15.         }
    16.     }, ctx);
    17.     new CountDownLatch(1).await();
    18. }

    3、节点查询

    • 获取节点数据 
      1. public class ZKGetNodeData implements Watcher {
      2.     private ZooKeeper zooKeeper = null;
      3.     private static final Logger log = LoggerFactory.getLogger(ZKGetNodeData.class);
      4.     private static final String zkServerPath = "127.0.0.1:2181";
      5.     private static final Integer timeout = 5000;
      6.  
      7.     public ZKGetNodeData() {}
      8.     public ZKGetNodeData(String connectString) {
      9.         try {
      10.             zooKeeper = new ZooKeeper(connectString, timeout, new ZKGetNodeData());
      11.         } catch (Exception e) {
      12.             e.printStackTrace();
      13.             if (zooKeeper != null) {
      14.                 try {
      15.                     zooKeeper.close();
      16.                 } catch (Exception e1) {
      17.                     e1.printStackTrace();
      18.                 }
      19.             }
      20.         }
      21.     }
      22.  
      23.     private static CountDownLatch countDownLatch = new CountDownLatch(1);
      24.     private static Stat stat = new Stat();
      25.     public static void main(String[] args) throws Exception {
      26.         ZKGetNodeData zkGetNodeData = new ZKGetNodeData(zkServerPath);
      27.         zkGetNodeData.getZooKeeper().create("/testnode","testnode".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
      28.         Thread.sleep(5000);
      29.         // 第一个参数: 节点path; 第二个参数: true注册一个监听事件; 第三个参数: 获取的结果会保存在stat
      30.         byte[] result = zkGetNodeData.getZooKeeper().getData("/testnode", true, stat);
      31.         log.warn("当前值: {}", new String(result));
      32.         countDownLatch.await();
      33.     }
      34.  
      35.     @Override
      36.     public void process(WatchedEvent event) {
      37.         try {
      38.             if (event.getType() == Event.EventType.NodeDataChanged) {
      39.                 ZKGetNodeData zkGetNodeData = new ZKGetNodeData(zkServerPath);
      40.                 byte[] result = zkGetNodeData.getZooKeeper().getData("/testnode", false, stat);
      41.                 log.warn("监听到值已经更改, 更改后的值为: {}, 版本号: {}", new String(result), stat.getVersion());
      42.                 countDownLatch.countDown(); // 计数器减1
      43.             } else if (event.getType() == Event.EventType.NodeCreated) {
      44.  
      45.             } else if (event.getType() == Event.EventType.NodeDeleted) {
      46.  
      47.             } else if (event.getType() == Event.EventType.NodeChildrenChanged) {
      48.  
      49.             }
      50.         } catch (Exception e) {
      51.             e.printStackTrace();
      52.         }
      53.     }
      54.  
      55.     public ZooKeeper getZooKeeper() {
      56.         return zooKeeper;
      57.     }
      58.  
      59.     public void setZooKeeper(ZooKeeper zooKeeper) {
      60.         this.zooKeeper = zooKeeper;
      61.     }
      62. }
    • 获取子节点列表 
      1. public class ZKGetChildrenList implements Watcher {
      2.     private ZooKeeper zooKeeper = null;
      3.     private static final Logger log = LoggerFactory.getLogger(ZKGetChildrenList.class);
      4.     private static final String zkServerPath = "127.0.0.1:2181";
      5.     private static final Integer timeout = 5000;
      6.  
      7.     public ZKGetChildrenList() {}
      8.     public ZKGetChildrenList(String connectString) {
      9.         try {
      10.             zooKeeper = new ZooKeeper(connectString, timeout, new ZKGetChildrenList());
      11.         } catch (Exception e) {
      12.             e.printStackTrace();
      13.             if (zooKeeper != null) {
      14.                 try {
      15.                     zooKeeper.close();
      16.                 } catch (Exception e1) {
      17.                     e1.printStackTrace();
      18.                 }
      19.             }
      20.         }
      21.     }
      22.  
      23.     private static CountDownLatch countDownLatch = new CountDownLatch(1);
      24.     private static Stat stat = new Stat();
      25.     public static void main(String[] args) throws Exception {
      26.         ZKGetChildrenList zkGetChildrenList = new ZKGetChildrenList(zkServerPath);
      27.         zkGetChildrenList.getZooKeeper().create("/zookeeper/bbb","bbb".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
      28.         Thread.sleep(5000);
      29.         // 同步调用: 参数1 节点路径, 参数2 true或false, 注册一个watch事件
      30.         List children = zkGetChildrenList.getZooKeeper().getChildren("/zookeeper", true);
      31.         for (String child : children) {
      32.             log.warn(child);
      33.         }
      34.  
      35.         // 异步调用
      36. //        String ctx = "{'callback':'ChildrenCallback'}";
      37. //        zkGetChildrenList.getZooKeeper().getChildren("/testnode", true, new AsyncCallback.ChildrenCallback() {
      38. //            @Override
      39. //            public void processResult(int rc, String path, Object ctx, List children) {
      40. //                log.warn("callback, path: {}, children: {}", path, children.toString());
      41. //            }
      42. //        }, ctx);
      43.         countDownLatch.await();
      44.     }
      45.  
      46.     @Override
      47.     public void process(WatchedEvent event) {
      48.         try {
      49.             if (event.getType() == Event.EventType.NodeDataChanged) {
      50.             } else if (event.getType() == Event.EventType.NodeCreated) {
      51.             } else if (event.getType() == Event.EventType.NodeDeleted) {
      52.             } else if (event.getType() == Event.EventType.NodeChildrenChanged) {
      53.                 ZKGetChildrenList zkGetChildrenList = new ZKGetChildrenList(zkServerPath);
      54.                 List children = zkGetChildrenList.getZooKeeper().getChildren("/zookeeper", false);
      55.                 log.warn("监听到子节点改变, 改变后子节点数组为:");
      56.                 for (String child : children) {
      57.                     log.warn(child);
      58.                 }
      59.                 countDownLatch.countDown(); // 计数器减1
      60.             }
      61.         } catch (Exception e) {
      62.             e.printStackTrace();
      63.         }
      64.     }
      65.  
      66.     public ZooKeeper getZooKeeper() {
      67.         return zooKeeper;
      68.     }
      69.  
      70.     public void setZooKeeper(ZooKeeper zooKeeper) {
      71.         this.zooKeeper = zooKeeper;
      72.     }
      73. }

    4、判断节点是否存在

    1. public class ZKNodeExist implements Watcher {
    2.     private ZooKeeper zooKeeper = null;
    3.     private static final Logger log = LoggerFactory.getLogger(ZKNodeExist.class);
    4.     private static final String zkServerPath = "127.0.0.1:2181";
    5.     private static final Integer timeout = 5000;
    6.  
    7.     public ZKNodeExist() {}
    8.     public ZKNodeExist(String connectString) {
    9.         try {
    10.             zooKeeper = new ZooKeeper(connectString, timeout, new ZKNodeExist());
    11.         } catch (Exception e) {
    12.             e.printStackTrace();
    13.             if (zooKeeper != null) {
    14.                 try {
    15.                     zooKeeper.close();
    16.                 } catch (Exception e1) {
    17.                     e1.printStackTrace();
    18.                 }
    19.             }
    20.         }
    21.     }
    22.  
    23.     private static CountDownLatch countDownLatch = new CountDownLatch(1);
    24.     public static void main(String[] args) throws Exception {
    25.         ZKNodeExist zkNodeExist = new ZKNodeExist(zkServerPath);
    26.         Stat stat = zkNodeExist.getZooKeeper().exists("/testnode", true);
    27.         if (stat == null) {
    28.             log.warn("节点/testnode不存在");
    29.         } else {
    30.             log.warn("节点/testnode存在. stat: {}", stat);
    31.         }
    32.  
    33.         countDownLatch.await();
    34.     }
    35.  
    36.     @Override
    37.     public void process(WatchedEvent event) {
    38.         try {
    39.             if (event.getType() == Event.EventType.NodeDataChanged) {
    40.             } else if (event.getType() == Event.EventType.NodeCreated) {
    41.             } else if (event.getType() == Event.EventType.NodeDeleted) {
    42.             } else if (event.getType() == Event.EventType.NodeChildrenChanged) {
    43.             }
    44.             countDownLatch.countDown();
    45.         } catch (Exception e) {
    46.             e.printStackTrace();
    47.         }
    48.     }
    49.  
    50.     public ZooKeeper getZooKeeper() {
    51.         return zooKeeper;
    52.     }
    53.  
    54.     public void setZooKeeper(ZooKeeper zooKeeper) {
    55.         this.zooKeeper = zooKeeper;
    56.     }
    57. }

    2.4 Watcher机制及ACL

    zookeeper有watch事件，是一次性触发的。当watch监视的数据发生变化时，通知在创建zookeeper是设置了Watcher的客户端。Watcher类监视的事件类型和状态类型如下所示：

    事件类型（znode节点相关）：

    • EventType.NodeCreated：节点创建

    • EventType.NodeDataChanged：节点数据变更

    • EventType.NodeChildrenChanged：子节点变更

    • EventType.NodeDeleted：节点删除

    状态类型（客户端实例相关）：

    • KeeperState.Disconnected：未连接

    • KeeperState.SyncConnected：已连接

    • KeeperState.AuthFailed：认证失败

    • KeeperState.Expired：会话失效

    Watcher的特性：一次性、客户端串行执行、轻量。

    • 一次性：对于ZK的Watcher，只需要记住一点：Zookeeper的watch事件是一次性触发的。当watch监视的数据发生变化时，通知设置了该watch的客户端，即watcher。由于zookeeper的监视都是一次性的，所以每次必须设置监控。

    • 客户端串行执行：客户端Watcher回调的过程是一个串行同步的过程，这为我们保证了顺序，同时需要注意一点，千万不要因为一个Watcher的处理逻辑影响了这个客户端的Watcher回调。

    • 轻量：WatchedEvent是Zookeeper整个Wacher通知机制的最小通知单元，整个数据结构只包含三部分：通知状态、事件类型和节点路径。也就是说Watcher通知非常的简单，只会告诉客户端发生了事件而不会告知其具体内容，需要客户端自己去获取，比如NodeDataChanged事件，Zookeeper只会通知客户端指定节点的数据发生了变更，而不会直接提供具体的数据内容。

    ACL（Access Control List），Zookeeper作为一个分布式协调框架，其内部存储的都是一些关乎分布式系统运行时状态的元数据，尤其是涉及到一些分布式锁、Master选举和协调等应用场景。我们需要有效的保障Zookeeper中的数据安全，Zookeeper提供了一套完善的ACL权限控制机制来保障数据的安全。

    Zookeeper提供了三种模式，权限模式、授权对象、权限：

    权限模式：Scheme，开发人员经常使用如下四种权限模式：

    • IP：ip模式通过ip地址粒度来进行权限控制，例如配置了：ip:192.168.1.107，即表示权限控制都是针对这个ip地址的，同时也支持按网段分配，比如：192.168.1.*。

    • Digest：digest是最常用的权限控制模式，也更符合对权限的认知。其类似于“username:password”形式的权限控制标识进行权限配置。Zookeeper会对形成的权限标识先后进行两次编码处理，分别是SHA-1加密算法和BASE64编码。

    • World：World是一种最开放的权限控制模式。这种模式可以看做为特殊的digest，它仅仅是一个标识而已。

    • Super：超级用户模式。在超级用户模式下可以对Zookeeper进行任意操作。

    权限对象：指的是权限赋予给用户或者一个指定的实体，例如IP地址或机器等。在不同的模式下，授权对象是不同的。这种模式和授权对象一一对应。

    权限：权限就是指那些通过权限检测后可以被允许执行的操作，在Zookeeper中，对数据的操作权限分为以下五大类：

    CREATE、DELETE、READ、WRITE、ADMIN

    1、自定义用户权限

    1. public class ZkNodeAcl implements Watcher {
    2.     private ZooKeeper zooKeeper = null;
    3.     private static final Logger log = LoggerFactory.getLogger(ZkNodeAcl.class);
    4.     private static final String zkServerPath = "127.0.0.1:2181";
    5.     private static final Integer timeout = 5000;
    6.  
    7.     public ZkNodeAcl() {}
    8.     public ZkNodeAcl(String connectString) {
    9.         try {
    10.             zooKeeper = new ZooKeeper(connectString, timeout, new ZkNodeAcl());
    11.         } catch (Exception e) {
    12.             e.printStackTrace();
    13.             if (zooKeeper != null) {
    14.                 try {
    15.                     zooKeeper.close();
    16.                 } catch (Exception e1) {
    17.                     e1.printStackTrace();
    18.                 }
    19.             }
    20.         }
    21.     }
    22.  
    23.     public static void main(String[] args) throws InterruptedException, NoSuchAlgorithmException, KeeperException {
    24.         ZkNodeAcl zkNodeOperator = new ZkNodeAcl(zkServerPath);
    25.         ArrayList acls = new ArrayList();
    26.         Id test1 = new Id("digest", DigestAuthenticationProvider.generateDigest("test1:123456"));
    27.         Id test2 = new Id("digest", DigestAuthenticationProvider.generateDigest("test2:123456"));
    28.         acls.add(new ACL(ZooDefs.Perms.ALL,test1));
    29.         acls.add(new ACL(ZooDefs.Perms.READ,test2));
    30.         acls.add(new ACL(ZooDefs.Perms.DELETE | ZooDefs.Perms.CREATE,test2));
    31.  
    32.         zkNodeOperator.createZKNode("/testacl", "heihei".getBytes(), acls);
    33.  
    34.         zkNodeOperator.getZooKeeper().addAuthInfo("digest", "test2:123456".getBytes());
    35.         Thread.sleep(10000);
    36.         Stat stat = new Stat();
    37.         byte[] result = zkNodeOperator.getZooKeeper().getData("/testacl", false, stat);
    38.         log.warn("当前值: {}, 版本: {}", new String(result), stat.getVersion());
    39.  
    40.         new CountDownLatch(1).await();
    41.     }
    42.  
    43.     private void createZKNode(String path, byte[] data, ArrayList acls) {
    44.         String result = "";
    45.         try {
    46.             String ctx = "{'create':'success'}";
    47.             zooKeeper.create(path, data, acls, CreateMode.PERSISTENT, new CreateCallBack(), ctx);
    48.             Thread.sleep(5000);
    49.             log.warn("异步创建节点: {} 成功。。。", result);
    50.  
    51.         } catch (Exception e) {
    52.             e.printStackTrace();
    53.         }
    54.     }
    55.  
    56.     @Override
    57.     public void process(WatchedEvent event) {
    58.         log.warn("客户端连接接收到watch通知： {}", event);
    59.     }
    60.  
    61.     public ZooKeeper getZooKeeper() {
    62.         return zooKeeper;
    63.     }
    64.  
    65.     public void setZooKeeper(ZooKeeper zooKeeper) {
    66.         this.zooKeeper = zooKeeper;
    67.     }
    68.  
    69. }

    2、acl之ip权限

    1. public class ZkNodeAclIp implements Watcher {
    2.     private ZooKeeper zooKeeper = null;
    3.     private static final Logger log = LoggerFactory.getLogger(ZkNodeAclIp.class);
    4.     private static final String zkServerPath = "127.0.0.1:2181";
    5.     private static final Integer timeout = 5000;
    6.  
    7.     public ZkNodeAclIp() {}
    8.     public ZkNodeAclIp(String connectString) {
    9.         try {
    10.             zooKeeper = new ZooKeeper(connectString, timeout, new ZkNodeAclIp());
    11.         } catch (Exception e) {
    12.             e.printStackTrace();
    13.             if (zooKeeper != null) {
    14.                 try {
    15.                     zooKeeper.close();
    16.                 } catch (Exception e1) {
    17.                     e1.printStackTrace();
    18.                 }
    19.             }
    20.         }
    21.     }
    22.  
    23.     public static void main(String[] args) throws Exception {
    24.         ZkNodeAclIp zkNodeAcl = new ZkNodeAclIp(zkServerPath);
    25.  
    26.         // ip 方式的 acl
    27.         ArrayList aclsIP = new ArrayList<>();
    28.         Id ipId1 = new Id("ip", "127.0.0.1");
    29.         aclsIP.add(new ACL(ZooDefs.Perms.ALL, ipId1));
    30.  
    31.         // 创建节点
    32.         zkNodeAcl.createZKNode("/testaclip", "testaclip".getBytes(), aclsIP);
    33.  
    34.         // 验证ip是否有权限
    35.         Stat stat = new Stat();
    36.         byte[] result = zkNodeAcl.getZooKeeper().getData("/testaclip", false, stat);
    37.         log.warn("当前值: {}, 版本: {}", new String(result), stat.getVersion());
    38.     }
    39.  
    40.     /**
    41.      * 创建节点
    42.      * @param path
    43.      * @param data
    44.      * @param acls
    45.      */
    46.     private void createZKNode(String path, byte[] data, ArrayList acls) {
    47.         String result = "";
    48.         try {
    49.             // 同步创建
    50.             result = zooKeeper.create(path, data, acls, CreateMode.PERSISTENT);
    51.             log.warn("同步创建临时节点: {} 成功。。。", result);
    52.         } catch (Exception e) {
    53.             e.printStackTrace();
    54.         }
    55.     }
    56.  
    57.     @Override
    58.     public void process(WatchedEvent event) {
    59.         log.warn("接收到watch通知： {}", event);
    60.     }
    61.  
    62.     public ZooKeeper getZooKeeper() {
    63.         return zooKeeper;
    64.     }
    65.  
    66.     public void setZooKeeper(ZooKeeper zooKeeper) {
    67.         this.zooKeeper = zooKeeper;
    68.     }
    69. }

    第3章 Curator应用

    Curator是netflix公司开源的一套zookeeper客户端，目前是Apache的顶级项目。与Zookeeper提供的原生客户端相比，Curator的抽象层次更高，简化了Zookeeper客户端的开发量。Curator解决了很多zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册wathcer和NodeExistsException 异常等

    引包

    1. <dependency>
    2.     <groupId>org.apache.zookeepergroupId>
    3.     <artifactId>zookeeperartifactId>
    4.     <version>3.4.13version>
    5. dependency>
    6. <dependency>
    7.     <groupId>org.apache.curatorgroupId>
    8.     <artifactId>curator-frameworkartifactId>
    9.     <version>4.0.1version>
    10.     <exclusions>
    11.         <exclusion>
    12.             <groupId>org.apache.zookeepergroupId>
    13.             <artifactId>zookeeperartifactId>
    14.         exclusion>
    15.     exclusions>
    16. dependency>
    17. <dependency>
    18.     <groupId>org.apache.curatorgroupId>
    19.     <artifactId>curator-recipesartifactId>
    20.     <version>4.0.1version>
    21.     <exclusions>
    22.         <exclusion>
    23.             <groupId>org.apache.zookeepergroupId>
    24.             <artifactId>zookeeperartifactId>
    25.         exclusion>
    26.     exclusions>
    27. dependency>
    28. dependencies>

    3.1 基础API

    1. public class CuratorBase {
    2.  
    3.     private static final Logger log = LoggerFactory.getLogger(CuratorBase.class);
    4.  
    5.     //zk服务地址
    6.     static final String zk_path = "127.0.0.1:2181";
    7.     //会话超时，默认60秒
    8.     static final int session_timeout=60000;
    9.     //连接超时时间
    10.     static final int connect_timeout=15000;
    11.  
    12.     /**
    13.      * 创建客户端
    14.      * @return
    15.      */
    16.     private static CuratorFramework createClient(){
    17.         //重连策略：1秒3次
    18.         RetryPolicy retryPolicy = new RetryNTimes(1000,3);
    19.         CuratorFramework zkClient = CuratorFrameworkFactory.builder()
    20.                 .connectionTimeoutMs(connect_timeout)
    21.                 .sessionTimeoutMs(session_timeout)
    22.                 .connectString(zk_path)
    23.                 .retryPolicy(retryPolicy)
    24.                 .build();
    25.         //开启链接
    26.         zkClient.start();
    27.         return zkClient;
    28.     }
    29.  
    30.     public static void baseAPI() throws Exception {
    31.         CuratorFramework zkCli = createClient();
    32.         CuratorFrameworkState state = zkCli.getState();
    33.         if(state.equals(CuratorFrameworkState.STARTED)){
    34.             /**
    35.              * 创建节点
    36.              *
    37.              * zk节点类型：
    38.              * PERSISTENT ： 持久化节点
    39.              * PERSISTENT_SEQUENTIAL ： 持久化有序节点
    40.              * EPHEMERAL ： 会话节点（伴随会话结束消失）
    41.              * EPHEMERAL_SEQUENTIAL ： 会话有序节点
    42.              */
    43.             String path = zkCli.create()
    44.                     .creatingParentsIfNeeded()
    45.                     .withMode(CreateMode.PERSISTENT)
    46.                     .forPath("/curator/base/1", "curator测试".getBytes());
    47.  
    48.             log.warn("path : {}",path);
    49.  
    50.             /**
    51.              * 获取节点数据
    52.              */
    53.             byte[] bytes = zkCli.getData().forPath(path);
    54.             log.warn("节点数据 : {} ",new String(bytes));
    55.  
    56.             /**
    57.              * 更新节点数据
    58.              */
    59.             zkCli.setData().forPath(path,"修改后的数据".getBytes());
    60.             byte[] bytes1 = zkCli.getData().forPath(path);
    61.             log.warn("更新节点数据 : {}",new String(bytes1));
    62.  
    63.             /**
    64.              * 获取子节点
    65.              */
    66.             List children_paths = zkCli.getChildren().forPath(path);
    67.             children_paths.forEach(x->{
    68.                 log.warn(path+" 子节点："+x);
    69.             });
    70.  
    71.             /**
    72.              * 检查节点状态
    73.              */
    74.             Stat stat = zkCli.checkExists().forPath(path);
    75.             log.warn(path+" 节点状态："+stat.toString());
    76.  
    77.             /**
    78.              * 删除节点
    79.              */
    80.             zkCli.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
    81.  
    82.             CountDownLatch countDownLatch = new CountDownLatch(1);
    83.             ExecutorService executorService = Executors.newCachedThreadPool();
    84.             String path2 = zkCli.create()
    85.                     .creatingParentsIfNeeded()
    86.                     .withMode(CreateMode.PERSISTENT)
    87.                     .inBackground(new BackgroundCallback() {
    88.                         @Override
    89.                         public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
    90.                             log.warn("code:" + curatorEvent.getResultCode());
    91.                             log.warn("type:" + curatorEvent.getType());
    92.                             log.warn("线程为:" + Thread.currentThread().getName());
    93.                             countDownLatch.countDown();
    94.                         }
    95.                     }, executorService)
    96.                     .forPath("/curator/base/2","curator测试2".getBytes());
    97.             countDownLatch.await();
    98.             if(path2!=null){
    99.                 byte[] bytes2 = zkCli.getData().forPath(path2);
    100.                 log.warn("/curator/base/2  :  "+ new String(bytes));
    101.             }
    102.         }
    103.     }
    104.  
    105.     public static void main(String[] args) throws Exception {
    106.         baseAPI();
    107.     }
    108.  
    109. }
    1. public class BaseOperator {
    2.     public static CuratorFramework getClient() {
    3.         return CuratorFrameworkFactory.builder()
    4.                 .connectString("127.0.0.1:2181")
    5.                 .retryPolicy(new ExponentialBackoffRetry(1000, 3))
    6.                 .connectionTimeoutMs(15 * 1000) //连接超时时间，默认15秒
    7.                 .sessionTimeoutMs(60 * 1000) //会话超时时间，默认60秒
    8.                 .namespace("arch") //设置命名空间
    9.                 .build();
    10.     }
    11.  
    12.     public static void create(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
    13.         client.create().creatingParentsIfNeeded().forPath(path, payload);
    14.     }
    15.  
    16.     public static void createEphemeral(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
    17.         client.create().withMode(CreateMode.EPHEMERAL).forPath(path, payload);
    18.     }
    19.  
    20.     public static String createEphemeralSequential(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
    21.         return client.create().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, payload);
    22.     }
    23.  
    24.     public static void setData(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
    25.         client.setData().forPath(path, payload);
    26.     }
    27.  
    28.     public static void delete(final CuratorFramework client, final String path) throws Exception {
    29.         client.delete().deletingChildrenIfNeeded().forPath(path);
    30.     }
    31.  
    32.     public static void guaranteedDelete(final CuratorFramework client, final String path) throws Exception {
    33.         client.delete().guaranteed().forPath(path);
    34.     }
    35.  
    36.     public static String getData(final CuratorFramework client, final String path) throws Exception {
    37.         return new String(client.getData().forPath(path));
    38.     }
    39.  
    40.     public static List getChildren(final CuratorFramework client, final String path) throws Exception {
    41.         return client.getChildren().forPath(path);
    42.     }
    43. }

    3.2 事件监听

    zookeeper原生支持通过注册watcher来进行事件监听，但是其使用不是特别方便，需要开发人员自己反复注册watcher，比较繁琐。

    Curator引入Cache来实现对zookeeper服务端事务的监听。Cache是Curator中对事件监听的包装，其对事件的监听其实可以近似看作是一个本地缓存视图和远程Zookeeper视图的对比过程。同时，Curator能够自动为开发人员处理反复注册监听，从而大大简化原生api开发的繁琐过程。

    1、NodeCache

    1. public class ZkCuratorNodeCache {
    2.     public static void main(String[] args) throws Exception {
    3.         nodeCache();
    4.     }
    5.  
    6.     public static void nodeCache() throws Exception {
    7.         final String path = "/nodeCache";
    8.         final CuratorFramework client = BaseOperator.getClient();
    9.         client.start();
    10.  
    11. //        BaseOperator.delete(client, path);
    12.         BaseOperator.create(client, path, "cache".getBytes());
    13.  
    14.         final NodeCache nodeCache = new NodeCache(client, path);
    15.         nodeCache.start(true);
    16.         nodeCache.getListenable()
    17.                 .addListener(() -> System.out.println("节点数据发生变化，新数据为：" + new String(nodeCache.getCurrentData().getData())));
    18.  
    19.         BaseOperator.setData(client, path, "cache1".getBytes());
    20.         BaseOperator.setData(client, path, "cache2".getBytes());
    21.  
    22.         Thread.sleep(1000);
    23.  
    24.         client.close();
    25.     }
    26. }

    NodeCache可以监听指定的节点，注册监听器后，节点的变化会通知相应的监听器

    2、Path Cache

    Path Cache 用来监听ZNode的子节点事件，包括added、updateed、removed，Path Cache会同步子节点的状态，产生的事件会传递给注册的PathChildrenCacheListener。

    1. public class ZkCuratorPathCache {
    2.     public static void main(String[] args) throws Exception {
    3.         pathChildrenCache();
    4.     }
    5.     public static void pathChildrenCache() throws Exception {
    6.         final String path = "/pathChildrenCache";
    7.         final CuratorFramework client = BaseOperator.getClient();
    8.         client.start();
    9.  
    10.         final PathChildrenCache cache = new PathChildrenCache(client, path, true);
    11.         cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
    12.         cache.getListenable().addListener((client1, event) -> {
    13.             switch (event.getType()) {
    14.                 case CHILD_ADDED:
    15.                     System.out.println("CHILD_ADDED:" + event.getData().getPath());
    16.                     break;
    17.                 case CHILD_REMOVED:
    18.                     System.out.println("CHILD_REMOVED:" + event.getData().getPath());
    19.                     break;
    20.                 case CHILD_UPDATED:
    21.                     System.out.println("CHILD_UPDATED:" + event.getData().getPath());
    22.                     break;
    23.                 case CONNECTION_LOST:
    24.                     System.out.println("CONNECTION_LOST:" + event.getData().getPath());
    25.                     break;
    26.                 case CONNECTION_RECONNECTED:
    27.                     System.out.println("CONNECTION_RECONNECTED:" + event.getData().getPath());
    28.                     break;
    29.                 case CONNECTION_SUSPENDED:
    30.                     System.out.println("CONNECTION_SUSPENDED:" + event.getData().getPath());
    31.                     break;
    32.                 case INITIALIZED:
    33.                     System.out.println("INITIALIZED:" + event.getData().getPath());
    34.                     break;
    35.                 default:
    36.                     break;
    37.             }
    38.         });
    39.  
    40. //        client.create().withMode(CreateMode.PERSISTENT).forPath(path);
    41.         Thread.sleep(1000);
    42.  
    43.         client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
    44.         Thread.sleep(1000);
    45.  
    46.         client.delete().forPath(path + "/c1");
    47.         Thread.sleep(1000);
    48.  
    49.         client.delete().forPath(path); //监听节点本身的变化不会通知
    50.         Thread.sleep(1000);
    51.  
    52.         client.close();
    53.     }
    54. }

    3、TreeCache

    Path Cache和Node Cache的“合体”，监视路径下的创建、更新、删除事件，并缓存路径下所有孩子结点的数据。

    1. public class ZkCuratorTreeCache {
    2.     public static void main(String[] args) throws Exception {
    3.         treeCache();
    4.     }
    5.  
    6.     public static void treeCache() throws Exception {
    7.         final String path = "/treeChildrenCache";
    8.         final CuratorFramework client = BaseOperator.getClient();
    9.         client.start();
    10.  
    11.         final TreeCache cache = new TreeCache(client, path);
    12.         cache.start();
    13.  
    14.         cache.getListenable().addListener((client1, event) -> {
    15.             switch (event.getType()){
    16.                 case NODE_ADDED:
    17.                     System.out.println("NODE_ADDED:" + event.getData().getPath());
    18.                     break;
    19.                 case NODE_REMOVED:
    20.                     System.out.println("NODE_REMOVED:" + event.getData().getPath());
    21.                     break;
    22.                 case NODE_UPDATED:
    23.                     System.out.println("NODE_UPDATED:" + event.getData().getPath());
    24.                     break;
    25.                 case CONNECTION_LOST:
    26.                     System.out.println("CONNECTION_LOST:" + event.getData().getPath());
    27.                     break;
    28.                 case CONNECTION_RECONNECTED:
    29.                     System.out.println("CONNECTION_RECONNECTED:" + event.getData().getPath());
    30.                     break;
    31.                 case CONNECTION_SUSPENDED:
    32.                     System.out.println("CONNECTION_SUSPENDED:" + event.getData().getPath());
    33.                     break;
    34.                 case INITIALIZED:
    35.                     System.out.println("INITIALIZED:" + event.getData().getPath());
    36.                     break;
    37.                 default:
    38.                     break;
    39.             }
    40.         });
    41.  
    42.         client.create().withMode(CreateMode.PERSISTENT).forPath(path);
    43.         Thread.sleep(1000);
    44.  
    45.         client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
    46.         Thread.sleep(1000);
    47.  
    48.         BaseOperator.setData(client, path, "test".getBytes());
    49.         Thread.sleep(1000);
    50.  
    51.         client.delete().forPath(path + "/c1");
    52.         Thread.sleep(1000);
    53.  
    54.         client.delete().forPath(path);
    55.         Thread.sleep(1000);
    56.  
    57.         client.close();
    58.     }
    59. }

    3.3 分布式锁应用

    1、可重入锁Shared Reentrant Lock

    Shared意味着锁是全局可见的， 客户端都可以请求锁。Reentrant和JDK的ReentrantLock类似， 意味着同一个客户端在拥有锁的同时，可以多次获取，不会被阻塞。它是由类InterProcessMutex来实现。它的构造函数为：

    public InterProcessMutex(CuratorFramework client, String path)

    2、不可重入锁Shared Lock

    使用InterProcessSemaphoreMutex，调用方法类似，区别在于该锁是不可重入的，在同一个线程中不可重入

    3、可重入读写锁Shared Reentrant Read Write Lock 类似JDK的ReentrantReadWriteLock. 一个读写锁管理一对相关的锁。一个负责读操作，另外一个负责写操作。读操作在写锁没被使用时可同时由多个进程使用，而写锁使用时不允许读 (阻塞)。此锁是可重入的。一个拥有写锁的线程可重入读锁，但是读锁却不能进入写锁。这也意味着写锁可以降级成读锁， 比如请求写锁 —>读锁 —->释放写锁。从读锁升级成写锁是不成的。主要由两个类实现：

    1. InterProcessReadWriteLock
    2. InterProcessLock

    4、信号量Shared Semaphore 一个计数的信号量类似JDK的Semaphore。JDK中Semaphore维护的一组许可(permits)，而Cubator中称之为租约(Lease)。注意，所有的实例必须使用相同的numberOfLeases值。调用acquire会返回一个租约对象。客户端必须在finally中close这些租约对象，否则这些租约会丢失掉。但是， 但是，如果客户端session由于某种原因比如crash丢掉， 那么这些客户端持有的租约会自动close， 这样其它客户端可以继续使用这些租约。租约还可以通过下面的方式返还：

    1. public void returnAll(Collection leases)
    2. public void returnLease(Lease lease)

    5、多锁对象Multi Shared Lock Multi Shared Lock是一个锁的容器。当调用acquire， 所有的锁都会被acquire，如果请求失败，所有的锁都会被release。同样调用release时所有的锁都被release(失败被忽略)。基本上，它就是组锁的代表，在它上面的请求释放操作都会传递给它包含的所有的锁。主要涉及两个类：

    1. InterProcessMultiLock
    2. InterProcessLock

    它的构造函数需要包含的锁的集合，或者一组ZooKeeper的path。

    1. public InterProcessMultiLock(List locks)
    2. public InterProcessMultiLock(CuratorFramework client, List paths)

    6、代码演示

    1. public class ZkCuratorLock {
    2.     private static final String zk_server = "127.0.0.1:2181";
    3.     private static final String zk_path = "/curator/zklock";
    4.  
    5.     public static void doWithLock(CuratorFramework curatorFramework){
    6.         List zkPaths = new ArrayList();
    7.         zkPaths.add(zk_path);
    8.         InterProcessMultiLock lock = new InterProcessMultiLock(curatorFramework,zkPaths);
    9. //        InterProcessMutex lock2 = new InterProcessMutex(curatorFramework,zk_path);
    10.         try {
    11.             if(lock.acquire(30, TimeUnit.SECONDS)){
    12.                 long threadId = Thread.currentThread().getId();
    13.                 System.out.println("线程-"+threadId+",acquire lock");
    14.                 Thread.sleep(1000);
    15.                 System.out.println("线程-"+threadId+",replease lock");
    16.             }
    17.         }catch (Exception e){
    18.             e.fillInStackTrace();
    19.         }finally {
    20.             try {
    21.                 lock.release();
    22.             } catch (Exception e) {
    23.                 e.fillInStackTrace();
    24.             }
    25.         }
    26.     }
    27.  
    28.     public static void main(String[] args) {
    29.         ExecutorService es = Executors.newFixedThreadPool(10);
    30.         for(int i=10;i>0;i--){
    31.             es.execute(new Runnable() {
    32.                 @Override
    33.                 public void run() {
    34.                     RetryNTimes retryNTimes = new RetryNTimes(1000, 3);
    35.                     CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(zk_server, retryNTimes);
    36.                     curatorFramework.start();
    37.                     ZkCuratorLock.doWithLock(curatorFramework);
    38.                 }
    39.             });
    40.         }
    41.         es.shutdown();
    42.     }
    43. }

    3.4 栅栏barrier

    1）DistributedBarrier构造函数中barrierPath参数用来确定一个栅栏，只要barrierPath参数相同(路径相同)就是同一个栅栏。通常情况下栅栏的使用如下：

    • 主导client设置一个栅栏

    • 其他客户端就会调用waitOnBarrier()等待栅栏移除，程序处理线程阻塞

    • 主导client移除栅栏，其他客户端的处理程序就会同时继续运行。 DistributedBarrier类的主要方法如下： setBarrier() - 设置栅栏 waitOnBarrier() - 等待栅栏移除 removeBarrier() - 移除栅栏

    2）双栅栏Double Barrier 双栅栏允许客户端在计算的开始和结束时同步。当足够的进程加入到双栅栏时，进程开始计算，当计算完成时，离开栅栏。双栅栏类是DistributedDoubleBarrier DistributedDoubleBarrier类实现了双栅栏的功能。它的构造函数如下：

    1. // client - the client
    2. // barrierPath - path to use
    3. // memberQty - the number of members in the barrier
    4. public DistributedDoubleBarrier(CuratorFramework client, String barrierPath, int memberQty)

    memberQty是成员数量，当enter方法被调用时，成员被阻塞，直到所有的成员都调用了enter。当leave方法被调用时，它也阻塞调用线程，直到所有的成员都调用了leave。 注意：参数memberQty的值只是一个阈值，而不是一个限制值。当等待栅栏的数量大于或等于这个值栅栏就会打开！ 与栅栏(DistributedBarrier)一样,双栅栏的barrierPath参数也是用来确定是否是同一个栅栏的，双栅栏的使用情况如下：

    • 从多个客户端在同一个路径上创建双栅栏(DistributedDoubleBarrier),然后调用enter()方法，等待栅栏数量达到memberQty时就可以进入栅栏。

    • 栅栏数量达到memberQty，多个客户端同时停止阻塞继续运行，直到执行leave()方法，等待memberQty个数量的栅栏同时阻塞到leave()方法中。

    • memberQty个数量的栅栏同时阻塞到leave()方法中，多个客户端的leave()方法停止阻塞，继续运行。 DistributedDoubleBarrier类的主要方法如下：enter()、enter(long maxWait, TimeUnit unit) - 等待同时进入栅栏 leave()、leave(long maxWait, TimeUnit unit) - 等待同时离开栅栏 异常处理：DistributedDoubleBarrier会监控连接状态，当连接断掉时enter()和leave方法会抛出异常。

    3.5 计数器Counters

    利用ZooKeeper可以实现一个集群共享的计数器。只要使用相同的path就可以得到最新的计数器值， 这是由ZooKeeper的一致性保证的。Curator有两个计数器， 一个是用int来计数，一个用long来计数。

    1）SharedCount 这个类使用int类型来计数。主要涉及三个类。

    • SharedCount

    • SharedCountReader

    • SharedCountListener SharedCount代表计数器， 可以为它增加一个SharedCountListener，当计数器改变时此Listener可以监听到改变的事件，而SharedCountReader可以读取到最新的值， 包括字面值和带版本信息的值VersionedValue。

    2）DistributedAtomicLong 除了计数的范围比SharedCount大了之外， 它首先尝试使用乐观锁的方式设置计数器， 如果不成功(比如期间计数器已经被其它client更新了)， 它使用InterProcessMutex方式来更新计数值。此计数器有一系列的操作：

    • get(): 获取当前值

    • increment()：加一

    • decrement(): 减一

    • add()：增加特定的值

    • subtract(): 减去特定的值

    • trySet(): 尝试设置计数值

    • forceSet(): 强制设置计数值

    你必须检查返回结果的succeeded()， 它代表此操作是否成功。如果操作成功， preValue()代表操作前的值， postValue()代表操作后的值。

    3.6 选举

    curator提供了两种方式，分别是Leader Latch和Leader Election。

    • Leader Latch

      随机从候选着中选出一台作为leader，选中之后除非调用close()释放leadship，否则其他的后选择无法成为leader

      1. public class LeaderLatchTest {
      2.  
      3.     private static final String PATH = "/demo/leader";
      4.  
      5.     public static void main(String[] args) {
      6.         List latchList = new ArrayList<>();
      7.         List clients = new ArrayList<>();
      8.         try {
      9.             for (int i = 0; i < 10; i++) {
      10.                 CuratorFramework client = getClient();
      11.                 client.start();
      12.                 clients.add(client);
      13.  
      14.                 final LeaderLatch leaderLatch = new LeaderLatch(client, PATH, "client#" + i);
      15.                 leaderLatch.addListener(new LeaderLatchListener() {
      16.                     @Override
      17.                     public void isLeader() {
      18.                         System.out.println(leaderLatch.getId() + ":I am leader. I am doing jobs!");
      19.                     }
      20.  
      21.                     @Override
      22.                     public void notLeader() {
      23.                         System.out.println(leaderLatch.getId() + ":I am not leader. I will do nothing!");
      24.                     }
      25.                 });
      26.                 latchList.add(leaderLatch);
      27.                 leaderLatch.start();
      28.             }
      29.             Thread.sleep(1000 * 60);
      30.         } catch (Exception e) {
      31.             e.printStackTrace();
      32.         } finally {
      33.             for (CuratorFramework client : clients) {
      34.                 CloseableUtils.closeQuietly(client);
      35.             }
      36.  
      37.             for (LeaderLatch leaderLatch : latchList) {
      38.                 CloseableUtils.closeQuietly(leaderLatch);
      39.             }
      40.         }
      41.     }
      42.  
      43.     public static CuratorFramework getClient() {
      44.         return CuratorFrameworkFactory.builder()
      45.                 .connectString("127.0.0.1:2181")
      46.                 .retryPolicy(new ExponentialBackoffRetry(1000, 3))
      47.                 .connectionTimeoutMs(15 * 1000) //连接超时时间，默认15秒
      48.                 .sessionTimeoutMs(60 * 1000) //会话超时时间，默认60秒
      49.                 .namespace("arch") //设置命名空间
      50.                 .build();
      51.     }
      52.  
      53. }

    • Leader Election

      通过LeaderSelectorListener可以对领导权进行控制， 在适当的时候释放领导权，这样每个节点都有可能获得领导权。而LeaderLatch则一直持有leadership， 除非调用close方法，否则它不会释放领导权。

      1. public class LeaderSelectorTest {
      2.     private static final String PATH = "/demo/leader";
      3.  
      4.     public static void main(String[] args) {
      5.         List selectors = new ArrayList<>();
      6.         List clients = new ArrayList<>();
      7.         try {
      8.             for (int i = 0; i < 10; i++) {
      9.                 CuratorFramework client = getClient();
      10.                 client.start();
      11.                 clients.add(client);
      12.  
      13.                 final String name = "client#" + i;
      14.                 LeaderSelector leaderSelector = new LeaderSelector(client, PATH, new LeaderSelectorListenerAdapter() {
      15.                     @Override
      16.                     public void takeLeadership(CuratorFramework client) throws Exception {
      17.                         System.out.println(name + ":I am leader.");
      18.                         Thread.sleep(2000);
      19.                     }
      20.                 });
      21.  
      22.                 leaderSelector.autoRequeue();
      23.                 leaderSelector.start();
      24.                 selectors.add(leaderSelector);
      25.             }
      26.             Thread.sleep(Integer.MAX_VALUE);
      27.         } catch (Exception e) {
      28.             e.printStackTrace();
      29.         } finally {
      30.             for (CuratorFramework client : clients) {
      31.                 CloseableUtils.closeQuietly(client);
      32.             }
      33.  
      34.             for (LeaderSelector selector : selectors) {
      35.                 CloseableUtils.closeQuietly(selector);
      36.             }
      37.  
      38.         }
      39.     }
      40.  
      41.     public static CuratorFramework getClient() {
      42.         return CuratorFrameworkFactory.builder()
      43.                 .connectString("127.0.0.1:2181")
      44.                 .retryPolicy(new ExponentialBackoffRetry(1000, 3))
      45.                 .connectionTimeoutMs(15 * 1000) //连接超时时间，默认15秒
      46.                 .sessionTimeoutMs(60 * 1000) //会话超时时间，默认60秒
      47.                 .namespace("arch") //设置命名空间
      48.                 .build();
      49.     }
      50.  
      51. }

      至此Zookeeper的应用大体讲完了，在这里多说一句，技术的API不用去背，背也是背不住的，多使用就好了。

    下文预告

    1. 理清Zookeeper的Session创建、刷新和过期流程分析

    2. 明确Zookeeper的核心业务调用链

  • 相关阅读:
    Oracle 注释详解（--、/**/、rem）
    Linux TCP 单机优化
    手把手教你5种方法实现Java判断字符串是否为数字
    【Python Web】Flask框架（五）Bootstrap登录和后台管理案例
    Spring源码笔记之SpringIOC--（2）从BeanDefinition到Bean实例
    大数据，对于生活的改变
    李沐推荐的算法入门书分享，有动画图解、能运行、可讨论
    ​LeetCode解法汇总2605. 从两个数字数组里生成最小数字
    【Redis】几款redis可视化工具(推荐Another Redis Desktop Manager)
    软件接口安全设计规范及审计要点
  • 原文地址：https://blog.csdn.net/Eclipse_2019/article/details/126331912