zookeeper之ZkClient

在zookeeper的API使用中，咱们已经知道了原生API的使用，可是原生api的Watcher注册反复注册，session超时重连等功能，都比较繁琐，因此有了各自开源客户端的出现。

maven依赖

<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.11</version>
</dependency>

建立会话

before，建立会话用，后面不在贴这部分代码

private static ZkClient zkClient;

@Before
public void before() {
    zkClient = ZkClientConnect.getZkClient();
}

ZkClientConnect

public class ZkClientConnect {
    static final String CONNECT_STRING = "172.17.0.2:2181,172.17.0.3:2181,172.17.0.4:2181";

    public static ZkClient getZkClient() {
        // 第一个参数，是zookeeper集群各个节点的地址，多个地址用逗号隔开
        // 第二个参数，会话超时时间，毫秒为单位。
        ZkClient zkClient = new ZkClient(CONNECT_STRING, 5000);
        return zkClient;
    }
}

测试代码：

@Test
public void testConnect() {
    System.out.println("链接成功");
}

运行结果以下：

原生API会话的建立，是异步的，咱们以前用CountDownLatch来控制，ZkClient已经内部处理，转为同步了。另外代码看起来清爽了不少，ZkClient不须要咱们传入Watcher，而是经过Listener来实现对Watcher的监听，这个下面会讲。

增删改查

建立节点

从截图的方法来看，建立节点，也简化了很多，提供了临时节点、永久节点、多层级建立等方法。
测试代码：

@Test
public void testCreate() {
    zkClient.create("/node1", "node1_data", CreateMode.EPHEMERAL);
    System.out.println("建立node1节点成功");

    zkClient.createEphemeral("/node2");
    System.out.println("建立临时节点node2成功");

    zkClient.createEphemeralSequential("/node3", "node3_data");
    System.out.println("建立临时有序节点node2成功");

    zkClient.createPersistent("/node4", "node4_data");
    System.out.println("建立持久节点node4成功");

    zkClient.createPersistentSequential("/node5", "node5_data");
    System.out.println("建立持久有序节点node5成功");
    // 为true说明能够多层级建立
    zkClient.createPersistent("/node6/node6_1", true);
    System.out.println("建立持久有序节点node6成功");

    try {
        // 留时间截图临时节点
        TimeUnit.SECONDS.sleep(10);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

运行结果以下：

客户端查询结果以下：

[zk: localhost:2181(CONNECTED) 38] ls /
[node1, node2, node30000000021, node4, node50000000023, node6, zookeeper]
[zk: localhost:2181(CONNECTED) 39] ls /node6
[node6_1]

在原生API中，若是要建立多层级，就要一层一层的建立，建立的时候，还要判断是否已经建立节点，比较麻烦。用ZkClient就会方便不少。传入的数据，也是object，不须要传入byte[]，序列化的工做ZkClient帮咱们处理了，固然，咱们也能够本身定义序列化工具。

删除节点

测试代码：

@Test
public void testDelete() {
    zkClient.delete("/node4");
    System.out.println("/node4删除成功");
    zkClient.deleteRecursive("/node6");
    System.out.println("/node6及子节点删除成功");
}

运行结果以下：

客户端查询结果以下：

[zk: localhost:2181(CONNECTED) 46] ls /
[node50000000023, zookeeper]

用deleteRecursive遍历删除子节点，仍是很是方便的。

节点列表

测试代码：

@Test
public void testGetChildren(){
    List<String> children = zkClient.getChildren("/");
    System.out.println(children);
}

运行结果以下：

对比与原生API，少了Watcher。

获取数据

测试代码：

@Test
public void testReadData() {
    String data = zkClient.readData("/node50000000023");
    System.out.println(data);
}

运行结果以下：

获取到的结果，直接帮咱们反序列化。

更新数据

测试代码：

@Test
public void testWriteData() {
    zkClient.writeData("/node50000000023","node5_new_data");
    String data = zkClient.readData("/node50000000023");
    System.out.println(data);
}

运行结果以下：

节点是否存在

测试代码：

@Test
public void testExists() {
    System.out.println("node是否存在：" + zkClient.exists("/node"));
}

运行结果以下：

这边返回的是boolean型，原生API是返回Stat。

Listener

getChildren Listener

测试代码：

@Test
public void testGetChildren4Listener() throws InterruptedException {
    // 对/children的子节点监听
    zkClient.subscribeChildChanges("/children", new IZkChildListener() {
        // 第一个参数是父节点，第二个参数是子节点集合
        public void handleChildChange(String parentPath, List<String> list) throws Exception {
            System.out.println("handleChildChange--" + parentPath + ":" + list);
        }
    });

    System.out.println("--------------1--------------");
    zkClient.createPersistent("/children");
    TimeUnit.MILLISECONDS.sleep(200);
    System.out.println(zkClient.getChildren("/children"));

    System.out.println("--------------2--------------");
    zkClient.createEphemeral("/children/children_1");
    TimeUnit.MILLISECONDS.sleep(200);
    System.out.println(zkClient.getChildren("/children"));

    System.out.println("--------------3--------------");
    zkClient.writeData("/children/children_1","children_1");
    TimeUnit.MILLISECONDS.sleep(200);
    System.out.println(zkClient.getChildren("/children"));
    zkClient.writeData("/children","children");
    TimeUnit.MILLISECONDS.sleep(200);

    System.out.println("--------------4--------------");
    zkClient.delete("/children/children_1");
    TimeUnit.MILLISECONDS.sleep(200);
    System.out.println(zkClient.getChildren("/children"));

    System.out.println("--------------5--------------");
    zkClient.delete("/children");
    TimeUnit.MILLISECONDS.sleep(200);
}

运行结果以下：

从一、5能够看出，当前节点的建立和删除，也会通知客户端。
从二、三、4能够看出，子节点的新增、删除，会通知客户端。
此外，不存在的节点也能够监听，并且不像Watcher，须要反复注册，ZkClient会一直监听。

DataChanges Listener

测试代码

@Test
public void testDataChanges4Listener() throws InterruptedException {
    zkClient.subscribeDataChanges("/node", new IZkDataListener() {
        public void handleDataChange(String dataPath, Object data) throws Exception {
            System.out.println("handleDataChange--" + dataPath + ":" + data);
        }

        public void handleDataDeleted(String dataPath) throws Exception {
            System.out.println("handleDataDeleted--" + dataPath);
        }
    });

    zkClient.createEphemeral("/node");
    TimeUnit.MILLISECONDS.sleep(200);
    zkClient.writeData("/node", "data");
    TimeUnit.MILLISECONDS.sleep(200);
    zkClient.delete("/node");
    TimeUnit.MILLISECONDS.sleep(200);
}

运行结果以下：
建立、修改的时候，调用的是handleDataChange方法，删除的时候调用的是handleDataDeleted方法