• Zookeeper的api使用

            Zookeeper作为⼀个分布式框架，主要用来解决分布式⼀致性问题，它提供了简单的分布式原语，并且对多种编程语⾔提供了API，所以接下来重点来看下Zookeeper的java客户端API使用方式

    Zookeeper API共包含五个包，分别为：

    1. org.apache.zookeeper
    2. org.apache.zookeeper.data
    3. org.apache.zookeeper.server
    4. org.apache.zookeeper.server.quorum
    5. org.apache.zookeeper.server.upgrade

            其中org.apache.zookeeper，包含Zookeeper类，他是我们编程时最常⽤的类文件。这个类是Zookeeper客户端的主要类文件。如果要使用Zookeeper服务，应⽤程序⾸先必须创建⼀个Zookeeper 实例，这时就需要使用此类。⼀旦客户端和Zookeeper服务端建立起了连接，Zookeeper系统将会给本次连接会话分配⼀个ID值，并且客户端将会周期性的向服务器端发送心跳来维持会话连接。只要连接有效，客户端就可以使⽤Zookeeper API来做相应处理了

    准备工作：导入依赖

    1. <dependency>
    2.     <groupId>org.apache.zookeepergroupId>
    3.     <artifactId>zookeeperartifactId>
    4.     <version>3.4.14version>
    5. dependency>

     

    建立会话

    1. package com.lagou.api;
    2.  
    3. import org.apache.zookeeper.WatchedEvent;
    4. import org.apache.zookeeper.Watcher;
    5. import org.apache.zookeeper.ZooKeeper;
    6.  
    7. import java.io.IOException;
    8. import java.util.concurrent.CountDownLatch;
    9.  
    10. public class CreateSession implements Watcher {
    11.     private static CountDownLatch countDownLatch = new CountDownLatch(1);
    12.  
    13.     /**
    14.      * 建立会话
    15.      */
    16.     public static void main(String[] args) throws IOException, InterruptedException {
    17.         /*
    18.             客户端可以通过创建⼀个zk实例来连接zk服务器
    19.             new Zookeeper(connectString,sesssionTimeOut,Wather)
    20.             connectString: 连接地址：IP：端⼝
    21.             sesssionTimeOut：会话超时时间：单位毫秒
    22.             Wather：监听器(当特定事件触发监听时，zk会通过watcher通知到客户端)
    23.         */
    24.         ZooKeeper zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new CreateSession());
    25.         System.out.println(zooKeeper.getState());
    26.  
    27.         // 计数工具类 CountDownLatch : 不让main方法结束，让线程处于等待阻塞
    28.         countDownLatch.await();
    29.         System.out.println("客户端与服务端会话真正建立了");
    30.     }
    31.  
    32.     /**
    33.      * 回调方法：处理来自服务器端的watcher通知
    34.      */
    35.     // 当前类实现了Watcher接⼝，重写了process⽅法，该⽅法负责处理来⾃Zookeeper服务端的watcher通知，在收到服务端发送过来的SyncConnected事件之后，解除主程序在CountDownLatch上的等待阻塞，⾄此，会话创建完毕
    36.     public void process(WatchedEvent watchedEvent) {
    37.         //当连接创建了，服务端发送给客户端SyncConnected事件
    38.         if(watchedEvent.getState() == Event.KeeperState.SyncConnected) {
    39.             // 解除主程序在CountDownLatch上的等待阻塞
    40.             countDownLatch.countDown();
    41.         }
    42.     }
    43. }

            注意，ZooKeeper 客户端和服务端会话的建立是⼀个异步的过程，也就是说在程序中，构造⽅法会在处理完客户端初始化工作后立即返回，在⼤多数情况下，此时并没有真正建立好⼀个可用的会话，在会话的生命周期中处于“CONNECTING”的状态。当该会话真正创建完毕后ZooKeeper服务端会向会话对应的客户端发送⼀个事件通知，以告知客户端，客户端只有在获取这个通知之后，才算真正建立了会话。

    创建节点

    1. package com.lagou.api;
    2.  
    3. import org.apache.zookeeper.*;
    4.  
    5. import java.io.IOException;
    6. import java.util.concurrent.CountDownLatch;
    7.  
    8. public class CreateNote implements Watcher {
    9.     private static ZooKeeper zooKeeper;
    10.  
    11.     /**
    12.      * 建立会话
    13.      */
    14.     public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
    15.         /*
    16.             客户端可以通过创建⼀个zk实例来连接zk服务器
    17.             new Zookeeper(connectString,sesssionTimeOut,Wather)
    18.             connectString: 连接地址：IP：端⼝
    19.             sesssionTimeOut：会话超时时间：单位毫秒
    20.             Wather：监听器(当特定事件触发监听时，zk会通过watcher通知到客户端)
    21.         */
    22.         zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new CreateNote());
    23.         System.out.println(zooKeeper.getState());
    24.  
    25.         Thread.sleep(Integer.MAX_VALUE);
    26.  
    27.     }
    28.  
    29.     // 创建节点的方法
    30.     private static void createNoteSync() throws InterruptedException, KeeperException {
    31.         /**
    32.          *	path	：节点创建的路径
    33.          *	data[]	：节点创建要保存的数据，是个byte类型的
    34.          *	acl	：节点创建的权限信息(4种类型)
    35.          *	    ANYONE_ID_UNSAFE	: 表示任何⼈
    36.          *	    AUTH_IDS	：此ID仅可⽤于设置ACL。它将被客户机验证的ID替换。
    37.          *	    OPEN_ACL_UNSAFE	：这是⼀个完全开放的ACL(常⽤)--> world:anyone
    38.          *	    CREATOR_ALL_ACL	：此ACL授予创建者身份验证ID的所有权限
    39.          *	createMode	：创建节点的类型(4种类型)
    40.          *	    PERSISTENT：持久节点
    41.          *	    PERSISTENT_SEQUENTIAL：持久顺序节点
    42.          *	    EPHEMERAL：临时节点
    43.          *	    EPHEMERAL_SEQUENTIAL：临时顺序节点
    44.          String node = zookeeper.create(path,data,acl,createMode);
    45.          */
    46.         // 持久节点
    47.         String note_persistent = zooKeeper.create("/lg-persistent", "持久节点内容".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    48.  
    49.         // 临时节点
    50.         String note_ephemeral = zooKeeper.create("/lg-ephemeral", "临时节点内容".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    51.  
    52.         // 持久顺序节点
    53.         String note_sequential = zooKeeper.create("/lg-sequential", "持久顺序节点内容".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
    54.  
    55.         System.out.println("创建的持久节点: " + note_persistent);
    56.         System.out.println("创建的临时节点: " + note_ephemeral);
    57.         System.out.println("创建的持久顺序节点: " + note_sequential);
    58.  
    59.     }
    60.  
    61.     /**
    62.      * 回调方法：处理来自服务器端的watcher通知
    63.      */
    64.     public void process(WatchedEvent watchedEvent) {
    65.         // SyncConnected
    66.         if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
    67.             // 创建节点
    68.             try {
    69.                 createNoteSync();
    70.             } catch (Exception e) {
    71.                 throw new RuntimeException(e);
    72.             }
    73.         }
    74.     }
    75. }

    获取节点数据

    1. package com.lagou.api;
    2.  
    3. import org.apache.zookeeper.*;
    4.  
    5. import java.io.IOException;
    6. import java.util.List;
    7.  
    8. public class GetNoteData implements Watcher {
    9.     private static ZooKeeper zooKeeper;
    10.  
    11.     /**
    12.      * 建立会话
    13.      */
    14.     public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
    15.         /*
    16.             客户端可以通过创建⼀个zk实例来连接zk服务器
    17.             new Zookeeper(connectString,sesssionTimeOut,Wather)
    18.             connectString: 连接地址：IP：端⼝
    19.             sesssionTimeOut：会话超时时间：单位毫秒
    20.             Wather：监听器(当特定事件触发监听时，zk会通过watcher通知到客户端)
    21.         */
    22.         zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new GetNoteData());
    23.         System.out.println(zooKeeper.getState());
    24.  
    25.         Thread.sleep(Integer.MAX_VALUE);
    26.  
    27.     }
    28.  
    29.  
    30.     /**
    31.      * 回调方法：处理来自服务器端的watcher通知
    32.      */
    33.     public void process(WatchedEvent watchedEvent) {
    34.         /*
    35.             子节点列表发生改变时，服务端会发送noteChildrenChanged事件通知
    36.             要重新获取子节点列表，同时注意：通知是一次性的，需要反复注册监听
    37.          */
    38.         if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged) {
    39.             List children = null;
    40.             try {
    41.                 children = zooKeeper.getChildren("/lg-persistent", true);
    42.             } catch (KeeperException | InterruptedException e) {
    43.                 throw new RuntimeException(e);
    44.             }
    45.             System.out.println(children);
    46.         }
    47.  
    48.         // SyncConnected
    49.         if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
    50.  
    51.             // 获取节点数据的方法
    52.             try {
    53.                 getNoteData();
    54.  
    55.                 // 获取节点的子节点列表方法
    56.                 getChildren();
    57.             } catch (Exception e) {
    58.                 throw new RuntimeException(e);
    59.             }
    60.         }
    61.     }
    62.  
    63.     /*
    64.         获取某个节点的内容
    65.      */
    66.     private void getNoteData() throws Exception {
    67.         /**
    68.          *	path	: 获取数据的路径
    69.          *	watch	: 是否开启监听
    70.          *	stat	: 节点状态信息
    71.          *	    null: 表示获取最新版本的数据
    72.          *	zk.getData(path, watch, stat);
    73.          */
    74.         byte[] data = zooKeeper.getData("/lg-persistent", false, null);
    75.         System.out.println(new String(data));
    76.     }
    77.  
    78.     /*
    79.         获取某个节点的子节点列表方法
    80.      */
    81.     public static void getChildren() throws InterruptedException, KeeperException {
    82.         /*
    83.             path:路径
    84.             watch:是否要启动监听，当⼦节点列表发⽣变化，会触发监听
    85.             zooKeeper.getChildren(path, watch);
    86.         */
    87.         List children = zooKeeper.getChildren("/lg-persistent", true);
    88.         System.out.println(children);
    89.     }
    90. }

    修改节点数据

    1. package com.lagou.api;
    2.  
    3. import org.apache.zookeeper.*;
    4. import org.apache.zookeeper.data.Stat;
    5.  
    6. import java.io.IOException;
    7.  
    8. public class UpdateNoteData implements Watcher {
    9.     private static ZooKeeper zooKeeper;
    10.  
    11.     /**
    12.      * 建立会话
    13.      */
    14.     public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
    15.         /*
    16.             客户端可以通过创建⼀个zk实例来连接zk服务器
    17.             new Zookeeper(connectString,sesssionTimeOut,Wather)
    18.             connectString: 连接地址：IP：端⼝
    19.             sesssionTimeOut：会话超时时间：单位毫秒
    20.             Wather：监听器(当特定事件触发监听时，zk会通过watcher通知到客户端)
    21.         */
    22.         zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new UpdateNoteData());
    23.         System.out.println(zooKeeper.getState());
    24.  
    25.         Thread.sleep(Integer.MAX_VALUE);
    26.  
    27.     }
    28.  
    29.  
    30.     /**
    31.      * 回调方法：处理来自服务器端的watcher通知
    32.      */
    33.     public void process(WatchedEvent watchedEvent) {
    34.         // SyncConnected
    35.         if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
    36.             // 更新数据节点内容的方法
    37.             try {
    38.                 updateNoteSync();
    39.             } catch (InterruptedException | KeeperException e) {
    40.                 throw new RuntimeException(e);
    41.             }
    42.         }
    43.     }
    44.  
    45.     /*
    46.         更新数据节点内容的方法
    47.      */
    48.     private void updateNoteSync() throws InterruptedException, KeeperException {
    49.         /*
    50.             path:路径
    51.             data:要修改的内容 byte[]
    52.             version:为-1，表示对最新版本的数据进⾏修改
    53.             zooKeeper.setData(path, data,version);
    54.         */
    55.         byte[] data = zooKeeper.getData("/lg-persistent", false, null);
    56.         System.out.println("修改前的值：" + new String(data));
    57.  
    58.         // 修改 /lg-persistent 的数据    stat: 状态信息对象
    59.         Stat stat = zooKeeper.setData("/lg-persistent", "客户端修改了节点数据".getBytes(), -1);
    60.         byte[] data2 = zooKeeper.getData("/lg-persistent", false, null);
    61.         System.out.println("修改后的值：" + new String(data2));
    62.     }
    63. }

    删除节点

    1. package com.lagou.api;
    2.  
    3. import org.apache.zookeeper.*;
    4. import org.apache.zookeeper.data.Stat;
    5.  
    6. import java.io.IOException;
    7.  
    8. public class DeleteNote implements Watcher {
    9.     private static ZooKeeper zooKeeper;
    10.  
    11.     /**
    12.      * 建立会话
    13.      */
    14.     public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
    15.         /*
    16.             客户端可以通过创建⼀个zk实例来连接zk服务器
    17.             new Zookeeper(connectString,sesssionTimeOut,Wather)
    18.             connectString: 连接地址：IP：端⼝
    19.             sesssionTimeOut：会话超时时间：单位毫秒
    20.             Wather：监听器(当特定事件触发监听时，zk会通过watcher通知到客户端)
    21.         */
    22.         zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new DeleteNote());
    23.         System.out.println(zooKeeper.getState());
    24.  
    25.         Thread.sleep(Integer.MAX_VALUE);
    26.  
    27.     }
    28.  
    29.  
    30.     /**
    31.      * 回调方法：处理来自服务器端的watcher通知
    32.      */
    33.     public void process(WatchedEvent watchedEvent) {
    34.         // SyncConnected
    35.         if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
    36.             // 删除节点
    37.             try {
    38.                 deleteNoteSync();
    39.             } catch (InterruptedException | KeeperException e) {
    40.                 throw new RuntimeException(e);
    41.             }
    42.         }
    43.     }
    44.  
    45.     /*
    46.         删除节点方法
    47.      */
    48.     private void deleteNoteSync() throws InterruptedException, KeeperException {
    49.         /*
    50.             zooKeeper.exists(path,watch) :判断节点是否存在
    51.             zookeeper.delete(path,version) : 删除节点
    52.         */
    53.         Stat stat = zooKeeper.exists("/lg-persistent/c1", false);
    54.         System.out.println(stat == null ? "该节点不存在" : "该节点存在");
    55.         if (stat != null) {
    56.             zooKeeper.delete("/lg-persistent/c1", -1);
    57.         }
    58.         Stat stat2 = zooKeeper.exists("/lg-persistent/c1", false);
    59.         System.out.println(stat2 == null ? "该节点不存在" : "该节点存在");
    60.     }
    61. }

  • 相关阅读:
    过滤器和监听器
    Linux常用命令
    JDBC---封装JDBC代码，配置properties文件
    阿里影业S1财报解读：优质内容叠加整合效益，转动增长飞轮
    懒人制作企业期刊的秘籍
    详解内存SDRAM原理(P-Bank、L-Bank、刷新、预充电等)
    谷粒商城 (二十七) --------- 商品服务 API 商品管理
    IDA pro逆向工具寻找socket server的IP和port
    SpringMVC基础：请求转发和重定向
    Explain执行计划字段解释说明---type字段说明（02）
  • 原文地址：https://blog.csdn.net/weixin_52851967/article/details/126259980