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方法