zookeeper客户端（二）zkClient的使用

- 日理万妓 2022-01-05 07:25 454阅读 0赞

### 原生Zookeeper API存在的弊端： ###

1.  Zookeeper API使用起来比较繁琐，并且zookeeper的watcher是一次性的，如果要基于watcher实现发布/订阅模式需要进行额外的编码，以实现每次watcher失效后重新注册，将一次性的订阅包装成持久订阅。
2.  Zookeeper API接口中，节点保存数据默认为二进制byte数组，若想直接保存对象类型的数据，需要将对象转换为二进制类型，因此还需要进行相关序列化等工作。

### zkClient的优点 ###

1.  zkClient解决了watcher一次性注册问题，将znode的事件重新定义为了子节点的变化，数据的变化及状态的变化三类，由zkClient统一将watcher的WatchedEvent转换为以上三种情况中去处理，watcher执行后重新读取数据的同时，再注册相同的watcher。
2.  zkClient在发送session expire异常时会自动创建新的zookeeper实例进行重连，由于原来所有的watcher和临时节点都会实现，因此可以在zkClient中定义的连接状态变化的接口IzkStateListener里面的handlerNewSession进行容错处理。
3.  zkClient提供了ZKserializer接口，可进行序列化和反序列化的相关操作，简化了znode的数据存储。

### zkClient的使用 ###

1、创建maven工程，在pom文件中引入zkClient。

<dependency>
        <groupId>com.github.sgroschupf</groupId>
        <artifactId>zkclient</artifactId>
        <version>0.1</version>
    </dependency>

2、API的使用

/**
     * 高级API ZkClient的使用, 实现了session超时自动重连、watcher反复注册等，减轻开发人员的负担
     */
    public class ZkClientDemo {
    
        private final String servers = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    
        private final int connectionTimeout = 2000; //ms
    
        private ZkClient zkClient = null;
    
        private final String path = "/zk-data";
    
        @Before
        public void initZK() {
            zkClient = new ZkClient(servers, connectionTimeout);
    
            //注册监听器，监听path下子节点数据的变化
            zkClient.subscribeDataChanges(path, new IZkDataListener() {
                public void handleDataChange(String dataPath, Object data) throws Exception {
                    System.out.println("handleDataChange, dataPath:" + dataPath + ", data:" + data);
                }
    
                public void handleDataDeleted(String dataPath) throws Exception {
                    System.out.println("handleDataDeleted, dataPath:" + dataPath);
                }
            });
    
            //注册监听器，监听path下子节点的变化
            zkClient.subscribeChildChanges(path, new IZkChildListener() {
                public void handleChildChange(String dataPath, List<String> children) throws Exception {
                    System.out.println("handleChildChange, " + "dataPath:" + dataPath + "->" + children);
                }
            });
        }
    
        @Test
        public void createNode() {
            //若节点已存在，则删除
            if(zkClient.exists(path)) {
               zkClient.delete(path);
            }
    
            //创建持久节点,并写入数据
            zkClient.createPersistent(path, "test-data");
        }
    
    
        @Test
        public void readData() {
            String data = zkClient.readData(path);
            System.out.println(data);
        }
    
        @Test
        public void updateData() {
            zkClient.writeData(path, "test-data1");
            try {
                Thread.sleep(Long.MAX_VALUE);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        //结果：handleDataChange, dataPath:/zk-data, data：test-data1
    
    
        @Test
        public void deleteNode() {
            zkClient.delete(path);
            try {
                Thread.sleep(Long.MAX_VALUE);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //结果：handleDataDeleted, dataPath:/zk-data
    }

zkClient节点有三种状态可供订阅，这种关系是永久的，而非一次性的。

**1、子节点的变化**

//注册监听器，监听path下子节点的变化
    zkClient.subscribeChildChanges(path, new IZkChildListener() {
          public void handleChildChange(String dataPath, List<String> children) throws Exception {
                System.out.println("handleChildChange, " + "dataPath:" + dataPath + "->" + children);
               }
    });

**2、数据的变化**

//注册监听器，监听path下子节点数据的变化
    zkClient.subscribeDataChanges(path, new IZkDataListener() {
                public void handleDataChange(String dataPath, Object data) throws Exception {
                    System.out.println("handleDataChange, dataPath:" + dataPath + ", data:" + data);
                }
    
                public void handleDataDeleted(String dataPath) throws Exception {
                    System.out.println("handleDataDeleted, dataPath:" + dataPath);
                }
    });

**3、状态的变化**

zkclient.subscribeStateChanges(new IZkStateListener(){
            		
         public void handleStateChanged(KeeperState state) throw Exception{
    
         }
    
         public void handleNewSession() throws Exception {
    
         }
    });

zkClient在发送session expire异常时会自动创建新的zookeeper实例进行重连，由于原来所有的watcher和临时节点都会实现，因此可以在zkClient中定义的连接状态变化的接口IzkStateListener里面的handlerNewSession进行容错处理。

文章参考：《大型分布式网站架构设计与实践》