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

关于zookeeper第三方客户端zkclient的使用说明_zk客户端运行依赖

作者：从前慢现在也慢 | 2024-02-16 15:36:59

赞

踩

zk客户端运行依赖

ZkClient

       在使用ZooKeeper的Java客户端时，经常需要处理几个问题：重复注册watcher、session失效重连、异常处理。

       要解决上述的几个问题，可以自己解决，也可以采用第三方的java客户端来完成。这里就介绍一种常用的客户端zkclient，目前已经运用到了很多项目中，知名的有Dubbo、Kafka、Helix。

 

ZKClient的设计

 

 

 

 

ZkClient的组件说明

从上述结构上看，IZKConnection是一个ZkClient与ZooKeeper之间的一个适配器。在代码里直接使用的是ZKClient，其实质还是委托了zookeeper来处理了。

       前面有一篇文章中，已经说了，使用ZooKeeper客户端来注册watcher有几种方法：1、创建ZooKeeper对象时指定默认的Watcher，2、getData()，3、exists()，4、getchildren。其中getdata,exists注册的是某个节点的事件处理器（watcher），getchildren注册的是子节点的事件处理器（watcher）。而在ZKClient中，根据事件类型，分为了节点事件（数据事件）、子节点事件。对应的事件处理器则是IZKDataListener和IZKChildListener。另外加入了Session相关的事件和事件处理器。

       ZkEventThread是专门用来处理事件的线程。

重要处理流程说明

 

启动ZKClient

在创建ZKClient对象时，就完成了到ZooKeeper服务器连接的建立。具体过程是这样的：

      

 

1、  启动时，指定好connection string，连接超时时间，序列化工具等。

2、  创建并启动eventThread，用于接收事件，并调度事件监听器Listener的执行。

3、  连接到zookeeper服务器，同时将ZKClient自身作为默认的Watcher。

 

为节点注册Watcher

       ZooKeeper的三个方法：getData、getChildren、exists，ZKClient都提供了相应的代理方法。就拿exists来看：

 

可以看到，是否注册watcher，由hasListeners(path)来决定的。

 

hasListeners就是看有没有与该数据节点绑定的listener。

 

所以呢，默认情况下，都会自动的为指定的path注册watcher，并且是默认的watcher（ZKClient）。怎么才能让hasListeners判定值为true呢，也就是怎么才能为path绑定Listener呢？

ZKClient提供了订阅功能：

 

 

一个新建的会话，只需要在取得响应的数据节点后，调用subscribteXxx就可以订阅上相应的事件了。

 

 

ZooKeeper的变更操作

Zookeeper中提供的变更操作有：节点的创建、删除，节点数据的修改。

 

创建操作，数据节点分为四种，ZKClient分别为他们提供了相应的代理：

 

删除节点的操作：

 

 

修改节点数据的操作：

 

 

writeDataReturnStat（）：写数据并返回数据的状态。

updateDataSerialized（）：修改已序列化的数据。执行过程是：先读取数据，然后使用DataUpdater对数据修改，最后调用writeData将修改后的数据发送给服务端。

 

客户端处理变更

       前面已经知道，ZKClient是默认的Watcher，并且在为各个数据节点注册的Watcher都是这个默认的Watcher。那么该是如何将各种事件通知给相应的Listener呢？

 

处理过程大致可以概括为下面的步骤：

1、判断变更类型：变更类型分为State变更、ChildNode变更（创建子节点、删除子节点、修改子节点数据）、NodeData变更（创建指定node，删除节点，节点数据变更）。

 

2、取出与path关联的Listeners，并为每一个Listener创建一个ZKEvent，将ZkEvent交给ZkEventThread处理。

3、ZkEventThread线程，拿到ZkEvent后，只需要调用ZkEvent的run方法进行处理。

 

从这里也可以知道，具体的怎么如何调用Listener，还要依赖于ZkEvent的run()实现了。

 

序列化处理

ZooKeeper中，会涉及到序列化、反序列化的操作有两种：getData、setData。在ZKClient中，分别用readData、writeData来替代了。

对于readData：先调用zookeeper的getData，然后进行使用ZKSerializer进行反序列化工作。

对于writeData：先使用ZKSerializer将对象序列化后，再调用zookeeper的setData。

 

ZkClient如何解决使用ZooKeeper客户端遇到的问题的呢？

 

Watcher自动重注册：这个要是依赖于hasListeners（）的判断，来决定是否再次注册。如果对此有不清晰的，可以看上面的流程处理的说明

       Session失效重连：如果发现会话过期，就先关闭已有连接，再重新建立连接。

       异常处理：对比ZooKeeper和ZKClient，就可以发现ZooKeeper的所有操作都是抛异常的，而ZKClient的所有操作，都不会抛异常的。在发生异常时，它或做日志，或返回空，或做相应的Listener调用。

 

 

相比于ZooKeeper官方客户端，使用ZKClient时，只需要关注实际的Listener实现即可。所以这个客户端，还是推荐大家使用的。

另外，是关于zkclient的一些接口，我们可以通过这些接口直接调用，使其完成一些相应的任务。

1.创建会话

[java]  view plain  copy
  1. public class createSession {  
  2.   
  3.     public static void main(String[] args) {  
  4.         //zk集群的地址  
  5.         String ZKServers = "192.168.30.164:2181,192.168.30.165:2181,192.168.30.166:2181";  
  6.       
  7.         /** 
  8.          * 创建会话 
  9.          * new SerializableSerializer() 创建序列化器接口，用来序列化和反序列化 
  10.          */  
  11.         ZkClient zkClient = new ZkClient(ZKServers,10000,10000,new SerializableSerializer());  
  12.           
  13.         System.out.println("conneted ok!");  
  14.           
  15.     }  
  16. }  

2.创建节点

[java]  view plain  copy
  1. public class createNode {  
  2.   
  3.     public static void main(String[] args) {  
  4.         //zk集群的地址  
  5.         String ZKServers = "192.168.30.164:2181,192.168.30.165:2181,192.168.30.166:2181";  
  6.         ZkClient zkClient = new ZkClient(ZKServers,10000,10000,new SerializableSerializer());  
  7.           
  8.         System.out.println("conneted ok!");  
  9.           
  10.         User user = new User();  
  11.         user.setId(1);  
  12.         user.setName("testUser");  
  13.           
  14.         /** 
  15.          * "/testUserNode" :节点的地址 
  16.          * user：数据的对象 
  17.          * CreateMode.PERSISTENT：创建的节点类型 
  18.          */  
  19.         String path = zkClient.create("/testUserNode", user, CreateMode.PERSISTENT);  
  20.         //输出创建节点的路径  
  21.         System.out.println("created path:"+path);  
  22.     }  
  23. }  

[java]  view plain  copy
  1. //注意：一定要实现序列化接口  implements Serializable  
  2. public class User implements Serializable{  
  3.       
  4.     private Integer id;  
  5.     private String name;  
  6.       
  7.     public Integer getId() {  
  8.         return id;  
  9.     }  
  10.     public void setId(Integer id) {  
  11.         this.id = id;  
  12.     }  
  13.     public String getName() {  
  14.         return name;  
  15.     }  
  16.     public void setName(String name) {  
  17.         this.name = name;  
  18.     }  
  19. }  

3.获取节点中的数据

[java]  view plain  copy
  1. public class getData {  
  2.     public static void main(String[] args) {  
  3.                 //zk集群的地址  
  4.                 String ZKServers = "192.168.30.164:2181,192.168.30.165:2181,192.168.30.166:2181";  
  5.                 ZkClient zkClient = new ZkClient(ZKServers,10000,10000,new SerializableSerializer());  
  6.                 System.out.println("conneted ok!");  
  7.                   
  8.                 Stat stat = new Stat();  
  9.                 //获取 节点中的对象  
  10.                 User  user = zkClient.readData("/testUserNode",stat);  
  11.                 System.out.println(user.getName());  
  12.                 System.out.println(stat);  
  13.     }  
  14. }  

4.判断节点是否存在

[java]  view plain  copy
  1. public class getData {  
  2.     public static void main(String[] args) {  
  3.                 //zk集群的地址  
  4.                 String ZKServers = "192.168.30.164:2181,192.168.30.165:2181,192.168.30.166:2181";  
  5.                 ZkClient zkClient = new ZkClient(ZKServers,10000,10000,new SerializableSerializer());  
  6.                 System.out.println("conneted ok!");  
  7.                   
  8.                 boolean e = zkClient.exists("/testUserNode");  
  9.                 //返回 true表示节点存在 ，false表示不存在  
  10.                 System.out.println(e);  
  11.     }  
  12. }  

5.删除节点

[java]  view plain  copy
  1. public class getData {  
  2.     public static void main(String[] args) {  
  3.                 //zk集群的地址  
  4.                 String ZKServers = "192.168.30.164:2181,192.168.30.165:2181,192.168.30.166:2181";  
  5.                 ZkClient zkClient = new ZkClient(ZKServers,10000,10000,new SerializableSerializer());  
  6.                 System.out.println("conneted ok!");  
  7.                   
  8.                 //删除单独一个节点，返回true表示成功  
  9.                 boolean e1 = zkClient.delete("/testUserNode");  
  10.                 //删除含有子节点的节点  
  11.                 boolean e2 = zkClient.deleteRecursive("/test");  
  12.                   
  13.                 //返回 true表示节点成功 ，false表示删除失败  
  14.                 System.out.println(e1);  
  15.     }  
  16. }  

6.更新数据

[java]  view plain  copy
  1. public static void main(String[] args) {  
  2.                 //zk集群的地址  
  3.                 String ZKServers = "192.168.30.164:2181,192.168.30.165:2181,192.168.30.166:2181";  
  4.                 ZkClient zkClient = new ZkClient(ZKServers,10000,10000,new SerializableSerializer());  
  5.                 System.out.println("conneted ok!");  
  6.                   
  7.                 User user = new User();  
  8.                 user.setId(2);  
  9.                 user.setName("testUser2");  
  10.                 /** 
  11.                  * testUserNode 节点的路径 
  12.                  * user 传入的数据对象 
  13.                  */  
  14.                 zkClient.writeData("/testUserNode", user);  
  15.     }  

7.订阅节点的信息改变（创建节点，删除节点，添加子节点）

[java]  view plain  copy
  1. public class SubscribeChildChanges {  
  2.     private static class ZKChildListener implements IZkChildListener{  
  3.         /** 
  4.          * handleChildChange： 用来处理服务器端发送过来的通知 
  5.          * parentPath：对应的父节点的路径 
  6.          * currentChilds：子节点的相对路径 
  7.          */  
  8.         public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {  
  9.               
  10.             System.out.println(parentPath);  
  11.             System.out.println(currentChilds.toString());  
  12.               
  13.         }  
  14.           
  15.     }  
  16.       
  17.     public static void main(String[] args) throws InterruptedException {  
  18.         //zk集群的地址  
  19.         String ZKServers = "192.168.30.164:2181,192.168.30.165:2181,192.168.30.166:2181";  
  20.         ZkClient zkClient = new ZkClient(ZKServers,10000,10000,new SerializableSerializer());  
  21.         System.out.println("conneted ok!");  
  22.         /** 
  23.          * "/testUserNode" 监听的节点，可以是现在存在的也可以是不存在的 
  24.          */  
  25.         zkClient.subscribeChildChanges("/testUserNode3", new ZKChildListener());  
  26.         Thread.sleep(Integer.MAX_VALUE);  
  27.     }  
  28. }  

8.订阅节点的数据内容的变化

[java]  view plain  copy
  1. public class SubscribeDataChanges {  
  2.     private static class ZKDataListener implements IZkDataListener{  
  3.   
  4.         public void handleDataChange(String dataPath, Object data) throws Exception {  
  5.               
  6.             System.out.println(dataPath+":"+data.toString());  
  7.         }  
  8.   
  9.         public void handleDataDeleted(String dataPath) throws Exception {  
  10.               
  11.             System.out.println(dataPath);  
  12.               
  13.         }  
  14.          
  15.           
  16.     }  
  17.       
  18.     public static void main(String[] args) throws InterruptedException {  
  19.         //zk集群的地址  
  20.         String ZKServers = "192.168.30.164:2181,192.168.30.165:2181,192.168.30.166:2181";  
  21.         ZkClient zkClient = new ZkClient(ZKServers,10000,10000,new SerializableSerializer());  
  22.         System.out.println("conneted ok!");  
  23.   
  24.         zkClient.subscribeDataChanges("/testUserNode", new ZKDataListener());  
  25.         Thread.sleep(Integer.MAX_VALUE);  
  26.           
  27.     }  
  28. }  





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

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

  

闽ICP备14008679号