ZkClient是由Datameer的工程師開發的開源客戶端，對Zookeeper的原生API進行了包裝，實現了超時重連、Watcher反覆註冊等功能。

ZKClient版本及原始碼

maven依賴

ZkClient目前有兩個不同artifactId的系列。 
其中最早的0.1版本maven依賴如下：

<dependency>
     <groupId>org.apache.zookeeper</groupId>
     <artifactId>zookeeper</artifactId>
     <version>3.4.9</version>
 </dependency>
 <dependency>
     <groupId>com.github.sgroschupf</groupId>
     <artifactId>zkclient</artifactId>
     <version>0.1</version>
 </dependency>
 

另外一個系列為的maven依賴為：

<dependency>
     <groupId>org.apache.zookeeper</groupId>
     <artifactId>zookeeper</artifactId>
     <version>3.4.9</version>
 </dependency>
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.10</version>
</dependency>
 

其中第二個系列包含了從0.1~0.10的版本。檢視dubbo的原始碼，我們可以看到，dubbo採用了第一個系列的0.1版本。

原始碼

ZkClient使用

以下我們以第二個系列的0.10版本為例來說明ZKClient的API和使用

建立會話

ZkClient提供了7中建立會話的方法：

public ZkClient(String serverstring)

public ZkClient(String zkServers, int connectionTimeout)

public ZkClient(String zkServers, int sessionTimeout, int connectionTimeout)

public ZkClient(String zkServers, int sessionTimeout, int connectionTimeout, ZkSerializer zkSerializer)

public ZkClient(final String zkServers, final int sessionTimeout, final int connectionTimeout, final ZkSerializer zkSerializer, final long operationRetryTimeout)

public ZkClient(IZkConnection connection)

public ZkClient(IZkConnection connection, int connectionTimeout)

public ZkClient(IZkConnection zkConnection, int connectionTimeout, ZkSerializer zkSerializer)

public ZkClient(final IZkConnection zkConnection, final int connectionTimeout, final ZkSerializer zkSerializer, final long operationRetryTimeout)
 

上面方法的引數如果我們熟悉原生API的話，不難理解其引數，基本上引數名都是自描述的。值得留意的是ZkClient將ZK原生API中的非同步處理進行了同步化。

其中一個引數IZkConnection是一個介面的定義。檢視介面的方法不難發現它是對ZK原生介面最直接的包裝。在此介面下面有兩個實現方法ZkConnection和InMemoryConnection。在日常中直接使用ZkConnection方法就可以解決大部分的常見業務需求。

引數ZkSerializer同樣是一個介面，定義了byte陣列序列化和反序列化的兩個方法。如果不傳遞此引數，則預設使用org.I0Itec.zkclient.serialize.SerializableSerializer實現類進行序列化。某些情況下此序列化介面會出現問題，比如亂碼。此時，開發者可以直接實現ZkSerializer介面，重寫自己的序列化方法。比如使用Hessian或Kryo等。

舉例：

String CONNECT_ADDR = "192.168.1.171:2181,192.168.1.172:2181,192.168.1.173:2181";  
ZkClient zkc = new ZkClient(new ZkConnection(CONNECT_ADDR), 10000);

建立節點

ZkClient提供了15個建立節點的方法：

public void createPersistent(String path)

public void createPersistent(String path, boolean createParents)

public void createPersistent(String path, boolean createParents, List<ACL> acl)

public void createPersistent(String path, Object data)

public void createPersistent(String path, Object data, List<ACL> acl)

public String createPersistentSequential(String path, Object data)

public String createPersistentSequential(String path, Object data, List<ACL> acl) 

public void createEphemeral(final String path)

public void createEphemeral(final String path, final List<ACL> acl)

public String create(final String path, Object data, final CreateMode mode)

public String create(final String path, Object data, final List<ACL> acl, final CreateMode mode) 

public void create(final String path, Object data, final CreateMode mode, final AsyncCallback.StringCallback callback, final Object context)

public void createEphemeral(final String path, final Object data)

public void createEphemeral(final String path, final Object data, final List<ACL> acl)

public String createEphemeralSequential(final String path, final Object data)

public String createEphemeralSequential(final String path, final Object data, final List<ACL> acl)

檢視原始碼可知，其實這些建立節點的方法都是對原生API的一層封裝而已，底層實現基本相同。值得留意的一點是，原生API的引數通過byte[]來傳遞節點內容，而ZkClient支援自定義序列化，因此可以傳輸Object物件。

 該API方法的引數說明如下表所示。

引數名	說明
path	指定資料節點的節點路徑，即API呼叫的目的是建立該節點
data	節點的初始資料內容，可以傳入null
mode	節點型別，是一個列舉型別，通常有4種可選的節點型別
acl	節點的ACL策略
callback	註冊一個非同步回撥函式
context	用於傳遞一個物件，可以在執行回撥函式的時候使用
createParents	指定是否建立父節點

示例：

zkc.createEphemeral("/temp");
// 可以支援遞迴的建立，但是不能遞迴賦值
zkc.createPersistent("/super/c1", true);

刪除節點

刪除節點提供了以下方法：

public boolean delete(final String path)

public boolean delete(final String path, final int version)

public boolean deleteRecursive(String path)

刪除API其實很簡單，重點說一下deleteRecursive介面，這個介面提供了遞迴刪除的功能。在原生API中，如果一個節點存在子節點，那麼它將無法直接刪除，必須一層層遍歷先刪除全部子節點，然後才能將目標節點刪除。

示例：

刪除 /temp節點
zkc.delete("/temp");
//遞迴刪除/super
zkc.deleteRecursive("/super");

讀取節點

獲取節點列表

public List<String> getChildren(String path)

此介面返回子節點的相對路徑列表。比如節點路徑為/test/a1和/test/a2，那麼當path為/test時，返回的結果為[a1,a2]。

獲取節點內容

public <T extends Object> T readData(String path)

public <T extends Object> T readData(String path, boolean returnNullIfPathNotExists)

public <T extends Object> T readData(String path, Stat stat)

通過方法返回引數的定義，就可以得知，返回的結果（節點的內容）已經被反序列化成物件了。

獲取節點並返回內容示例：

List<String> list = zkc.getChildren("/super");
for(String p : list){
    System.out.println(p);
    String rp = "/super/" + p;
    String data = zkc.readData(rp);
    System.out.println("節點為：" + rp + "，內容為: " + data);
}

更新資料

更新操作可以通過以下介面來實現：

public void writeData(String path, Object object)

public void writeData(final String path, Object datat, final int expectedVersion)

public Stat writeDataReturnStat(final String path, Object datat, final int expectedVersion)

示例：zkc.writeData("/super/c1", "新內容");

監測節點是否存在

此API比較簡單，呼叫以下方法即可：

protected boolean exists(final String path, final boolean watch)

綜合以上api 寫一個小栗子：

程式碼在 bjsxt.zkclient.base

public class ZkClientBase {

    /** zookeeper地址 */
    static final String CONNECT_ADDR = "192.168.1.31:2181,192.168.1.32:2181,192.168.1.33:2181";
    /** session超時時間 */
    static final int SESSION_OUTTIME = 10000;//ms


    public static void main(String[] args) throws Exception {
        ZkClient zkc = new ZkClient(new ZkConnection(CONNECT_ADDR), SESSION_OUTTIME);
        //1. create and delete方法

        zkc.createEphemeral("/temp");
        zkc.createPersistent("/super/c1", true);
        Thread.sleep(10000);
        zkc.delete("/temp");
        zkc.deleteRecursive("/super");

        //2. 設定path和data 並且讀取子節點和每個節點的內容
        zkc.createPersistent("/super", "1234");
        zkc.createPersistent("/super/c1", "c1內容");
        zkc.createPersistent("/super/c2", "c2內容");
        List<String> list = zkc.getChildren("/super");
        for(String p : list){
            System.out.println(p);
            String rp = "/super/" + p;
            String data = zkc.readData(rp);
            System.out.println("節點為：" + rp + "，內容為: " + data);
        }

        //3. 更新和判斷節點是否存在
        zkc.writeData("/super/c1", "新內容");
        System.out.println(zkc.readData("/super/c1").toString());
        System.out.println(zkc.exists("/super/c1"));

//      4.遞迴刪除/super內容
        zkc.deleteRecursive("/super");
    }
}

輸出

c1
節點為：/super/c1，內容為: c1內容
c2
節點為：/super/c2，內容為: c2內容
新內容
true

註冊監聽

我們發現，上述ZKClient裡面並沒有類似的watcher、watch引數，這也就是說我們開發人員無需關心反覆註冊watcher的問題， 在ZkClient中可以通過註冊相關的事件監聽來實現對Zookeeper服務端時間的訂閱。其中ZkClient提供的監聽事件介面有以下幾種：

其中ZkClient還提供了一個unsubscribeAll方法，來解除所有監聽。

示例：

1 subscribeChildChanges方法 訂閱子節點變化

引數1： path路徑 
引數2：實現了IZKChildListener介面的類（如：例項化IZKClientListener類） 只需要重寫其handleChildChanges(String parentPath, List currentChild)方法。其中 
- parentPath 為監聽節點全路徑 
- currentChilds為新的子節點列表 
IZKChildListener事件說明針對於下面三個事件觸發： 
- 新增子節點 
- 減少子節點 
- 刪除節點 
注意： 不監聽節點內容的變化 
舉例

public class ZkClientWatcher1 {

    /** zookeeper地址 */
    static final String CONNECT_ADDR = "192.168.1.31:2181,192.168.1.32:2181,192.168.1.33:2181";
    /** session超時時間 */
    static final int SESSION_OUTTIME = 10000;//ms


    public static void main(String[] args) throws Exception {
        ZkClient zkc = new ZkClient(new ZkConnection(CONNECT_ADDR), SESSION_OUTTIME);

        //對父節點新增監聽子節點變化。
        zkc.subscribeChildChanges("/super", new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                System.out.println("parentPath: " + parentPath);
                System.out.println("currentChilds: " + currentChilds);
            }
        });

        Thread.sleep(3000);

        zkc.createPersistent("/super");
        Thread.sleep(1000);

        zkc.createPersistent("/super" + "/" + "c1", "c1內容");
        Thread.sleep(1000);

        zkc.createPersistent("/super" + "/" + "c2", "c2內容");
        Thread.sleep(1000);     

        zkc.delete("/super/c2");
        Thread.sleep(1000); 

        zkc.deleteRecursive("/super");
        Thread.sleep(Integer.MAX_VALUE);

    }
}

輸出

parentPath: /super
currentChilds: []
parentPath: /super
currentChilds: [c1]
parentPath: /super
currentChilds: [c1, c2]
parentPath: /super
currentChilds: [c1]
parentPath: /super
currentChilds: null
parentPath: /super
currentChilds: null

• 2 subscribeDataChanges 訂閱內容變化

和前面的subscribeChildChanges類似 
- 引數1 路徑 
- 引數2 IZkDataListener物件，重寫handleDataDeleted(String path) 方法，可以得到刪除節點的path 重寫handleDataChange(String path, Object data)可以得到變更的節點和變更的內容

舉例

public class ZkClientWatcher2 {

    /** zookeeper地址 */
    static final String CONNECT_ADDR = "192.168.1.31:2181,192.168.1.32:2181,192.168.1.33:2181";
    /** session超時時間 */
    static final int SESSION_OUTTIME = 10000;//ms


    public static void main(String[] args) throws Exception {
        ZkClient zkc = new ZkClient(new ZkConnection(CONNECT_ADDR), SESSION_OUTTIME);

        zkc.createPersistent("/super", "1234");

        //對父節點新增監聽子節點變化。
        zkc.subscribeDataChanges("/super", new IZkDataListener() {
            @Override
            public void handleDataDeleted(String path) throws Exception {
                System.out.println("刪除的節點為:" + path);
            }

            @Override
            public void handleDataChange(String path, Object data) throws Exception {
                System.out.println("變更的節點為:" + path + ", 變更內容為:" + data);
            }
        });

        Thread.sleep(3000);
        zkc.writeData("/super", "456", -1);
        Thread.sleep(1000);

        zkc.delete("/super");
        Thread.sleep(Integer.MAX_VALUE);
    }
}

• 輸出

變更的節點為:/super, 變更內容為:456
刪除的節點為:/super

3 zookeeper服務狀態監聽器：

public class ZkStateWatcher {
    static final String CONNECT_ADDR = "172.21.121.53:2181,172.21.121.54:2181,172.21.121.55:2181";
    static final int CONNECTION_OUTTIME = 5000;
    public static void main(String[] args) throws InterruptedException{
        ZkClient zkc = new ZkClient(new ZkConnection(CONNECT_ADDR),CONNECTION_OUTTIME);
        zkc.subscribeStateChanges(new IZkStateListener() {
            
            @Override
            public void handleStateChanged(KeeperState state) throws Exception {
                if(state==KeeperState.SyncConnected){
                    //當我重新啟動後start，監聽觸發
                    System.out.println("連線成功");
                }else if(state==KeeperState.Disconnected){
                    System.out.println("連線斷開");//當我在服務端將zk服務stop時，監聽觸發
                }else
                    System.out.println("其他狀態"+state);
                
            }
            
            @Override
            public void handleNewSession() throws Exception {
                System.out.println("---->重建session");
            }
        });
//        zkc.close();
        
        Thread.sleep(Integer.MAX_VALUE);
    }
        
}
當我只連線zk服務叢集中單獨的一個zk服務時：

./zkServer.sh stop 連線斷開

./zkServer.sh start 連線成功

./zkServer.sh restart 連線斷開 連線成功

./zkServer.sh stop 連線斷開

等待幾分鐘

./zkServer.sh satrt 其他狀態Expired ---->重建session 連線成功

其實無需等待幾分鐘，在zoo.cfg配置檔案中有2個引數來管理session的超時時間：minSessionTimeout和maxSessionTimeout

2個引數的預設值都為-1，這個值在日誌中可以看到：

預設情況下，maxSessionTimeout = ticketTime*20; minSessionTimeout = ticketTime*2;