Zkclient操作zookeeper案例

水深无声 2022-02-28 05:20 504阅读 0赞

Zkclient操作zookeeper

每天多学一点点~
话不多说，这就开始吧…

文章目录

  • Zkclient操作zookeeper
    • 1.Zookeeper的Java客户端API
    • 2.代码
      • 2.1. pom依赖
      • 2.2. zkCilent的序列化
      • 2.3 zkCilent的Util
      • 2.4 zkCilent的测试
    • 3.结语

1.Zookeeper的Java客户端API

有三种方式

  1. 原生
    （1）Zookeeper的Watcher是一次性的，每次触发之后都需要重新进行注册；
    （2）Session超时之后没有实现重连机制；
    （3）异常处理繁琐，Zookeeper提供了很多异常，对于开发人员来说可能根本不知道该如何处理这些异常信息；
    （4）只提供了简单的byte[]数组的接口，没有提供针对对象级别的序列化；
    （5）创建节点时如果节点存在抛出异常，需要自行检查节点是否存在；
    （6）删除节点无法实现级联删除；
  2. zkclient
    ZkClient是一个开源客户端，在Zookeeper原生API接口的基础上进行了包装，更便于开发人员使用。内部实现了Session超时重连，Watcher反复注册等功能。像dubbo等框架对其也进行了集成使用。
    虽然ZkClient对原生API进行了封装，但也有它自身的不足之处：
    （1）几乎没有参考文档；
    （2）异常处理简化（抛出RuntimeException）；
    （3）重试机制比较难用；
    （4）没有提供各种使用场景的实现；
  3. Curator
    Curator是Netflix公司开源的一套Zookeeper客户端框架，和ZkClient一样，解决了非常底层的细节开发工作，包括连接重连、反复注册Watcher和NodeExistsException异常等。目前已经成为Apache的顶级项目。另外还提供了一套易用性和可读性更强的Fluent风格的客户端API框架。

本文选用zkclient 进行操作

2.代码

在这里插入图片描述

2.1. pom依赖

  1. <dependency>
  2.       <groupId>com.101tec</groupId>
  3.       <artifactId>zkclient</artifactId>
  4.       <version>0.10</version>
  5.     </dependency>

2.2. zkCilent的序列化

防止数据在客户端呈现乱码，且 能 启动监听机制

  1. package com.jiagouedu.zkclient.watcher;
  4. import org.I0Itec.zkclient.exception.ZkMarshallingError;
  5. import org.I0Itec.zkclient.serialize.ZkSerializer;
  7. import java.io.UnsupportedEncodingException;
  9. /**
  10.  * @Classname zk序列化 和 反 序列化 方式
  11.  * @Description TODO
  12.  * @Date 2019/3/24 3:54
  13.  * @Created by 爆裂无球
  14.  */
  15. public class MyZkSerializer implements ZkSerializer {
  16.     /**
  17.      *  zk自带的序列化
  18.      */
  19.     public Object deserialize(byte[] bytes) throws ZkMarshallingError {
  20.         try {
  21.             return new String(bytes, "UTF-8");
  22.         } catch (UnsupportedEncodingException e) {
  23.             e.printStackTrace();
  24.         }
  25.         return null;
  26.     }
  28.     public byte[] serialize(Object obj) throws ZkMarshallingError {
  29.         try {
  30.             return String.valueOf(obj).getBytes("UTF-8");
  31.         } catch (UnsupportedEncodingException e) {
  32.             e.printStackTrace();
  33.         }
  34.         return null;
  35.     }
  36. }

2.3 zkCilent的Util

  1. package com.jiagouedu.zkclient.znode;
  2. import com.jiagouedu.zkclient.watcher.MyZkSerializer;
  3. import org.I0Itec.zkclient.ZkClient;
  4. import org.apache.zookeeper.CreateMode;
  5. import org.apache.zookeeper.ZooDefs;
  7. import java.util.List;
  10. /**
  11.  * @Classname ZkClientCrud
  12.  * @Description TODO
  13.  * @Date 2019/3/24 2:54
  14.  * @Created by 爆裂无球
  15.  */
  16. public class ZkClientCrud<T> {
  17.  ZkClient zkClient;
  18.     //   private String connectString="192.168.0.31:2181,192.168.0.32:2181,192.168.0.33:2181";
  19.     private String connectString = "127.0.0.1:2181";
  21.     //   public ZkClientUtil() {
  22. //      this.zkClient = new ZkClient(connectString,5000,5000,new SerializableSerializer());  // zk默认的序列化方式
  23. //   }
  25.     public ZkClientUtil() {
  26.         //定义序列化方式，别忘了
  27.         this.zkClient = new ZkClient(connectString, 5000, 5000, new MyZkSerializer());
  28.     }
  30.     public void createPersistent(String path) {
  31.         zkClient.createPersistent(path, true); //创建持久化节点，true表示如果父节点不存在则创建父节点
  32.     }
  34.     //创建 永久 节点，并设置数据
  35.     public void createPersistent(String path, Object data) {
  36.         zkClient.createPersistent(path, data);
  37.     }
  39.     // 创建永久 有序节点
  40.     public void createPersistentSequential(String path, Object data) {
  41.         zkClient.createEphemeralSequential(path, data);
  42.     }
  44.     //创建临时节点  会话失效后删除
  45.     public void createEphemeral(String path, Object data) {
  46.         zkClient.createEphemeral(path, data);
  47.     }
  49.     //创建 临时节点 有序 节点   会话失效后删除
  50.     public void createEphemeralSequential(String path, Object data) {
  51.         zkClient.createEphemeralSequential(path, data);
  52.     }
  54.     //创建alc节点
  55.     public void createAcl(String path, Object data, final List<ACL> acl, final CreateMode mode) {
  56.         zkClient.create(path, data, acl, mode);
  57.     }
  59.     //设置acl 属性
  60.     public void setAcl(String path, List<ACL> acl) {
  61.         zkClient.setAcl(path, acl);
  62.     }
  64.     //获得acl属性
  65.     public Map.Entry<List<ACL>, Stat> getAcl(String path) {
  66.         return zkClient.getAcl(path);
  67.     }
  70.     //读取数据
  71.     public T readData(String path) {
  72. //      return zkClient.readData(path);
  73.         //没有不会抛异常，而是返回null
  74.         return zkClient.readData(path, true);
  75.     }
  77.     /**
  78.      * 读取 子节点 只能找 其 子一级 下 所有的
  79.      */
  80.     public List<String> getChildren(String path) {
  81.         return zkClient.getChildren(path);
  82.     }
  84.     /**
  85.      * 递归查找 所有 子节点
  86.      */
  87.     public void getChilderRecursive(String path) {
  88.         System.out.println(path);
  89.         if (zkClient.exists(path)) {
  90.             List<String> list = zkClient.getChildren(path);
  91.             if (list.size() > 0) {
  92.                 list.stream().forEach(n -> {
  93.                     getChilderRecursive(path + "/" + n);
  94.                 });
  95.             }
  96.         }
  98.     }
  100.     // 更新内容
  101.     public void writeData(String path, Object object) {
  102.         zkClient.writeData(path, object);
  103.     }
  105.     //删除单个节点 即这个节点下不能有子节点
  106.     public void delete(String path) {
  107.         zkClient.delete(path);
  108.     }
  111.     //递归删除节点 即删除其节点下 所有子节点  对应rmr 命令
  112.     public void deleteRecursive(String path) {
  113.         zkClient.deleteRecursive(path);
  114.     }
  117.     /***
  118.      * 支持创建递归方式 但是不能写入数据
  119.      * @param path
  120.      * @param createParents     true，表明需要递归创建父节点
  121.      */
  122.     public void createPersistentRecursive(String path, boolean createParents) {
  123.         zkClient.createPersistent(path, createParents);
  124.     }
  126.     /**
  127.      * 关闭zk
  128.      */
  129.     public void close() {
  130.         zkClient.close();
  131.     }
  132.   /**
  133.      * 监听
  134.      */
  135.     public void lister(String path) {
  136.         //对节点添加监听变化。  当前节点内容修改、节点删除 能监听数据
  137.         zkClient.subscribeDataChanges(path, new IZkDataListener() {
  138.             @Override
  139.             public void handleDataChange(String dataPath, Object data) throws Exception {
  140.                 System.out.printf("   变更的节点为:%s,%s", dataPath, data);       // 节点变更  变更的节点为:/w,123
  141.             }
  143.             @Override
  144.             public void handleDataDeleted(String dataPath) throws Exception {
  145.                 System.out.printf("    删除的节点为:%s", dataPath);
  146.             }
  147.         });
  148.         //对父节点添加监听子节点变化。监听 下面子节点的新增、删除 和 当前节点   不监听数据发生修改和变化。  parentPath: /w,currentChilds:[ww1]
  149.         zkClient.subscribeChildChanges(path, new IZkChildListener() {
  150.             @Override
  151.             public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
  152.                 System.out.println("   parentPath: " + parentPath + ",currentChilds:" + currentChilds);
  153.             }
  154.         });
  155.         //zeng  gai  shan
  156.         //对父节点添加监听子节点变化。
  157.         zkClient.subscribeStateChanges(new IZkStateListener() {
  158.             @Override
  159.             public void handleStateChanged(Watcher.Event.KeeperState state) throws Exception {
  160.                 if (state == Watcher.Event.KeeperState.SyncConnected) {
  161.                     //当我重新启动后start，监听触发
  162.                     System.out.println("连接成功");
  163.                 } else if (state == Watcher.Event.KeeperState.Disconnected) {
  164.                     System.out.println("连接断开");//当我在服务端将zk服务stop时，监听触发
  165.                 } else
  166.                     System.out.println("其他状态" + state);
  167.             }
  169.             @Override
  170.             public void handleNewSession() throws Exception {
  171.                 System.out.println("重建session");
  172.             }
  174.             @Override
  175.             public void handleSessionEstablishmentError(Throwable error) throws Exception {
  176.             }
  177.         });
  179.     }
  181. }

2.4 zkCilent的测试

  1. package com.jiagouedu.zkclient.ZkSerializer;
  2. import com.jiagouedu.zkclient.watcher.MyZkSerializer;
  3. import org.I0Itec.zkclient.IZkChildListener;
  4. import org.I0Itec.zkclient.IZkDataListener;
  5. import org.I0Itec.zkclient.IZkStateListener;
  6. import org.I0Itec.zkclient.ZkClient;
  7. import org.I0Itec.zkclient.serialize.SerializableSerializer;
  8. import org.apache.zookeeper.Watcher;
  10. import java.util.List;
  12. /**
  13.  * @Classname 监听
  14.  * @Description TODO
  15.  * @Date 2019/3/24 3:38
  16.  * @Created by 爆裂无球
  17.  */
  18. public class ZkClientWatcher<T> {
  19.      private ZkClientUtil zkClientUtil = new ZkClientUtil();
  20.     ZkClient zkClient;
  22.     /**
  23.      * 新增 节点(持节+临时) 并读取数据   (不能直接创建 子节点)
  24.      */
  25.     @Test
  26.     public void createPersistent() {
  27.         User user = new User();
  28.         user.setAge(18);
  29.         user.setName("zjq");
  30.         zkClientUtil.createPersistent("/ry", user);       // 创建永久节点
  31.         System.out.println(zkClientUtil.readData("/ry"));  // 读取 节点 数据
  32.         User user2 = new User();
  33.         user2.setAge(23);
  34.         user2.setName("lj");
  35.         zkClientUtil.createEphemeral("/rh", user2);        // 创建临时节点
  36.         System.out.println(zkClientUtil.readData("/rh"));
  37.     }
  39.     /**
  40.      * 新增 节点(永久有序，临时有序)   并读取数据   (不能直接创建 子节点)
  41.      */
  42.     @Test
  43.     public void create() {
  44.         User user = new User();
  45.         user.setAge(1);
  46.         user.setName("我擦");
  47.         zkClientUtil.createPersistentSequential("/ry", user);    // 创建 永久有序节点
  49.         User user2 = new User("我擦2", 2);
  50.         zkClientUtil.createPersistentSequential("/ry", user2);    // 创建 临时有序节点
  52.     }
  54.     /**
  55.      * 创建acl节点
  56.      */
  57.     @Test
  58.     public void createAcl() {
  59.         User user = new User("acldata", 3);
  60.         User user2 = new User("acldata2", 4);
  61.         List<ACL> aclList = new ArrayList<>();
  62.         int perm = ZooDefs.Perms.ADMIN | ZooDefs.Perms.READ | ZooDefs.Perms.WRITE;          // 或 运算 admin 可以设置节点访问控制列表权限
  63.         aclList.add(new ACL(perm, new Id("world", "anyone")));                //设置有所人的权限
  64. //        aclList.add(new ACL(ZooDefs.Perms.ALL, new Id("ip", "127.0.0.1")));               //设置所有权限，本机ip
  65.         zkClientUtil.createPersistent("/acl", user);                                   // 创建节点
  66.         zkClientUtil.setAcl("/acl", aclList);                                        // 给节点设置acl权限
  67.         //创建并设置acl节点 ZooDefs.Ids.OPEN_ACL_UNSAFE 表anyone
  68.         zkClientUtil.createAcl("/alc2", aclList, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  69.     }
  71.     /**
  72.      * 获得 acl 属性
  73.      */
  74.     @Test
  75.     public void getAndSetAcl() {
  76.         Map.Entry acl = zkClientUtil.getAcl("/acl");
  77.         System.out.println(acl.getKey());
  78.         System.out.println(acl.getValue());
  79.     }
  81.     /**
  82.      * 递归创建节点(但是不能写入数据) 写入数据，读数据,,再 删除子节点
  83.      */
  84.     @Test
  85.     public void test() {
  86.         String path = "/zjq/t1";
  87.         zkClientUtil.createPersistentRecursive(path, true);
  88.         zkClientUtil.writeData(path, new User("xx", 3));
  89.         Object o = zkClientUtil.readData(path);
  90.         System.out.println(o);
  91.         zkClientUtil.deleteRecursive(path);
  92.     }
  94.     /**
  95.      * 创建子节点，并读取
  96.      */
  97.     @Test
  98.     public void getChildren() {
  99.         String path = "/zjq/t1";
  100.         String path2 = "/zjq/t2";
  101.         String path3 = "/zjq/t3";
  102.         zkClientUtil.createPersistentRecursive(path, true);
  103.         zkClientUtil.createPersistentRecursive(path2, true);
  104.         zkClientUtil.createPersistentRecursive(path3, true);
  105.         List list = zkClientUtil.getChildren("/zjq");
  106.         list.stream().forEach(n -> {
  107.             System.out.println(n);
  108.         });
  109.     }
  111.     /**
  112.      * 递归查找 所有 子节点
  113.      */
  114.     @Test
  115.     public void getChilderRecursive() {
  116.         String path = "/zjq";
  117.         zkClientUtil.getChilderRecursive(path);
  118.     }
  121.     /**
  122.      *  测试监听     并且开启 下面的main方法
  123.      */
  124.     @Test
  125.     public void testListen() throws InterruptedException {
  126.         ZkClientUtil zkClientUtil = new ZkClientUtil();
  127.         String path = "/wukong/w1";
  128.         zkClientUtil.deleteRecursive(path);                     //先删除
  129.         zkClientUtil.lister(path);                              //添加监听
  130.         zkClientUtil.createPersistent(path, "123");      //再创建节点
  131.         Thread.sleep(2000);
  132.         zkClientUtil.writeData(path, "abc");            //修改数据
  133.         Thread.sleep(Integer.MAX_VALUE);
  134.     }
  136.     public static void main(String[] args) throws InterruptedException {
  137.         ZkClientUtil zkClientUtil=new ZkClientUtil();
  138.         String path="/wukong/w1";
  139.         zkClientUtil.writeData(path,"abc"); //能触发 或者在sh zkCli.sh  delete /wukong 也行
  140.     }
  141. }

大概内容就这么多，不全的可以查查详细文档

3.结语

Curator好比hibernate，而zkclient就好比mybatis，一般情况下，zkclient可以满足需求~
世上无难事，只怕有心人，每天积累一点点，fighting！！！

发表评论

表情：
评论列表 （有 0 条评论，504人围观）

还没有评论，来说两句吧...

相关阅读