前言

在真正的專案中通常使用的是zkclient和curator，而不是原生的zookeeper客戶端，因為zookeeper原生的客戶端存在一定的侷限性，本篇小編主要講解一下這兩種zookeeper客戶端的使用！

內容

1.1zk原生api不足之處:

• 超時重連，不支援自動，需要手動操作
• Watch註冊一次後失效
• 不支援遞迴建立節點

1.2zkclient

ZkClient是一個開源客戶端，在Zookeeper原生API介面的基礎上進行了包裝，更便於開發人員使用。內部實現了Session超時重連，Watcher反覆註冊等功能。像dubbo等框架對其也進行了整合使用。

雖然ZkClient對原生API進行了封裝，但也有它自身的不足之處：

• 幾乎沒有參考文件；
• 異常處理簡化（丟擲RuntimeException）；
• 重試機制比較難用；
• 沒有提供各種使用場景的實現；
  zkclient簡單程式碼示例：

package com.zk.dev.zkClient.day1;  

import org.I0Itec.zkclient.IZkDataListener;  
import org.I0Itec.zkclient.ZkClient;  
import org.junit.After;  
import org.junit.Before;  
import
 
 org.junit.Test;  

import java.util.concurrent.TimeUnit;  

public class ZKTest  {  

    private ZkClient zk;  

    private String nodeName = "/test";  

    @Before  
    public void initTest() {  
        zk = new ZkClient("localhost:2181");  
    }  

    @After  
    public void dispose() {  
        zk.close();  
    }  

    @Test
 
  
    public void testListener() throws InterruptedException {  
        // 監聽指定節點的資料變化  

        zk.subscribeDataChanges(nodeName, new IZkDataListener() {  
            public void handleDataChange(String s, Object o) throws Exception {  
                System.out.println("node data changed!");  
                 System.out.println("node=>" + s);  
                 System.out.println("data=>" + o);  
                 System.out.println("--------------");  
            }  

            public void handleDataDeleted(String s) throws Exception {  
                System.out.println("node data deleted!");  
                System.out.println("s=>" + s);  
                System.out.println("--------------");  

            }  
        });  

                System.out.println("ready!");  

        // junit測試時，防止執行緒退出  
        while (true) {  
            TimeUnit.SECONDS.sleep(5);  
        }  
    }  

    @Test  
    public void testUpdateConfig() throws InterruptedException {  
        if (!zk.exists(nodeName)) {  
        //建立永久
            zk.createPersistent(nodeName);  
        }  
        zk.writeData(nodeName, "1");  
        zk.writeData(nodeName, "2");  
        zk.delete(nodeName);  
        zk.delete(nodeName);   
        zk.writeData("/test/ba", "bbb");  
    }  
}  

1.3Apache curator

• Apache開源
• 解決watcher的註冊一次就失效
• Api更加簡單易用
• 提供跟多解決方案並且實現:比如分散式鎖
• 提供常用的zookeeper工具類
• 程式設計風格更爽

curator簡單程式碼示例：

package com.zk.dev.zkClient.day1;  

import org.apache.curator.RetryPolicy;  
import org.apache.curator.framework.CuratorFramework;  
import org.apache.curator.framework.CuratorFrameworkFactory;  
import org.apache.curator.framework.recipes.cache.NodeCache;  
import org.apache.curator.framework.recipes.cache.NodeCacheListener;  
import org.apache.curator.framework.recipes.cache.PathChildrenCache;  
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;  
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;  
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;  
import org.apache.curator.retry.ExponentialBackoffRetry;  
import org.apache.zookeeper.CreateMode;  
import org.apache.zookeeper.ZooDefs.Ids;  


public class CuratorUtils {  

    public String connectString = "localhost:2181";  
    CuratorFramework  zkclient = null ;  
    public CuratorUtils(){  
        /** 
         * connectString連線字串中間用分號隔開,sessionTimeoutMs session過期時間,connectionTimeoutMs連線超時時間,retryPolicyc連線重試策略 
         */  
        //CuratorFrameworkFactory.newClient(connectString, sessionTimeoutMs, connectionTimeoutMs, retryPolicy)  
        // fluent風格aip   
  //    CuratorFrameworkFactory.builder().sessionTimeoutMs(5000).connectString(connectString).namespace("/test").build();  
        // 重連策略,沒1一秒重試一次,最大重試次數3次  
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);  
        zkclient = CuratorFrameworkFactory.builder().connectString(connectString).sessionTimeoutMs(5000).retryPolicy(retryPolicy).namespace("tests").build();  
        zkclient.start();  
    }  
    /** 
     * 遞迴建立節點 
     * @param path 
     * @param data 
     * @throws Exception 
     */  
    public void createNode(String path, byte[] data) throws Exception{  
        zkclient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).withACL(Ids.OPEN_ACL_UNSAFE).forPath(path, data);  
    }  
    /** 
     * 遞迴刪除節點 
     * @param path 
     * @throws Exception 
     */  
    public void delNode(String path) throws Exception{  
        zkclient.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);  
    }   public void zkClose(){  
        zkclient.close();  
    }  
    public void delNodeCallBack(String path) throws Exception{  
        zkclient.delete().guaranteed().deletingChildrenIfNeeded().inBackground(new DeleteCallBack()).forPath(path);  
    }      
    public void dataChanges(String path) throws Exception{  
        final NodeCache  dataWatch =  new NodeCache(zkclient, path);  
        dataWatch.start(true);  
        dataWatch.getListenable().addListener(new NodeCacheListener(){  

            public void nodeChanged() throws Exception {  
                System.out.println("path==>" + dataWatch.getCurrentData().getPath() + "==data==>" + new String(dataWatch.getCurrentData().getData()));  
            }  

        });  
        zkclient.delete().guaranteed().deletingChildrenIfNeeded().inBackground(new DeleteCallBack()).forPath(path);  
    }      
    public void addChildWatcher(String path) throws Exception{  
        final PathChildrenCache pc = new PathChildrenCache(zkclient, path, true);  
        pc.start(StartMode.POST_INITIALIZED_EVENT);  
        System.out.println("節點個數===>" + pc.getCurrentData().size());  
        pc.getListenable().addListener(new  PathChildrenCacheListener() {  

            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {  
                System.out.println("事件監聽到"  + event.getData().getPath());  
                if(event.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED)){  
                    System.out.println("客戶端初始化節點完成"  + event.getData().getPath());  
                }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)){  
                    System.out.println("新增節點完成"  + event.getData().getPath());  
                }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){  
                    System.out.println("刪除節點完成"  + event.getData().getPath());  
                }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){  
                    System.out.println("修改節點完成"  + event.getData().getPath());  
                }  
            }  
        });  

    }  
    /*執行執行緒*/  
    public static void main(String[] args) throws Exception{  
        CuratorUtils cu = new CuratorUtils();  
        cu.zkclient.setData().forPath("/aa", "love is not".getBytes());  
        cu.addChildWatcher("/aa");  
        try{  
            Thread.sleep(20000000);  
        }catch(Exception e){};  
    }  
 }  

curator結合spring：

pom中引入：

 <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-framework</artifactId>
                <version>${curator.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-recipes</artifactId>
                <version>${curator.version}</version>
            </dependency>
            <!-- curator(zookeeper的客戶端)包 -->
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-client</artifactId>
                <version>${curator.version}</version>
            </dependency>

  <!--zookeeper重試策略-->
    <bean id="retryPolicy" class="org.apache.curator.retry.RetryNTimes">
      <!--重試次數-->
      <constructor-arg index="0" value="10"></constructor-arg>
     <!--每次的間隔-->
      <constructor-arg index="1" value="5000"></constructor-arg>
  </bean>

    <!--zookeeper客戶端-->
    <bean id="client" class="org.apache.curator.framework.CuratorFrameworkFactory" factory-method="newClient" init-method="start">
        <constructor-arg index="0" value="192.168.25.131:2181"/>
        <!--session timeout 會話超時時間-->
        <constructor-arg index="1" value="10000"/>
        <!--建立連線的超時時間-->
        <constructor-arg index="2" value="5000"/>
        <!--重試策略-->
        <constructor-arg index="3" ref="retryPolicy"/>
    </bean>

參考部落格