zookeeper的原生api相對來說比較繁瑣，比如：對節點新增監聽事件，當監聽觸發後，我們需要再次手動新增監聽，否則監聽只生效一次；再比如，斷線重連也需要我們手動程式碼來判斷處理等等。對於curator的介紹，從網上百度了一段：Curator是Netflix開源的一套zookeeper客戶端框架，用它來操作zookeeper更加方便，按Curator官方所比喻的，guava to JAVA，curator to zookeeper，Curator採用了fluent風格的程式碼，非常簡潔。

和Curator相比, 另一個ZooKeeper客戶端——zkClient(https://github.com/sgroschupf/zkclient

----------------------------------------------------------------------------------------------------------------------------------

　　Curator包含6部分，均可提供單獨jar包，每個包簡單介紹如下：

　　client：zk-client的替代品，提供一些底層處理跟工具類；

　　framework：高階封裝，大大簡化了zk的客戶端程式設計，包含對zk的連線管理，重試機制等；

　　repices：提供了一些常用的操作，比如持續監聽，鎖，選舉等；

　　utilities：各種工具類；

　　errors：curator對異常跟錯誤的處理；

　　extendsion：擴充套件包；

maven依賴如下

1. <dependency>
2.     <groupId>org.apache.curator</groupId>
3.     <artifactId>curator-recipes
   </artifactId>
4.     <version>2.5.0</version>
5. </dependency>

按照官方給出的文件和包結構，可以輕鬆的看出Curator功能分兩大類，一是對zookeeper的一些基本命令的封裝，比如增刪改查。是他的framework模組，一個是他的高階特性，即recipes模組。

一、framework模組

Curator提供了一套Fluent風格的操作API。這在很多指令碼類語言裡比較流行。

比如他建立client的程式碼是這樣

1. CuratorFramework client = builder.connectString("192.168.11.56:2180")  
2.         .sessionTimeoutMs(30000)  
3.         .connectionTimeoutMs(30000)  
4.         .canBeReadOnly(false)  
5.         .retryPolicy(new ExponentialBackoffRetry(1000, Integer.MAX_VALUE))  
6.         .namespace(namespace)  
7.         .defaultData(null)  
8.         .build();  
9. client.start();  

 一路點到底，這就是所謂的Fluent風格。 

我們再看增刪改查的

1. publicclass CrudExamples {  
2.     privatestatic CuratorFramework client = ClientFactory.newClient();  
3.     privatestaticfinal String PATH = "/crud";  
4.     publicstaticvoid main(String[] args) {  
5.         try {  
6.             client.start();  
7.             client.create().forPath(PATH, "I love messi".getBytes());  
8.             byte[] bs = client.getData().forPath(PATH);  
9.             System.out.println("新建的節點，data為:" + new String(bs));  
10.             client.setData().forPath(PATH, "I love football".getBytes());  
11.             // 由於是在background模式下獲取的data，此時的bs可能為null
12.             byte[] bs2 = client.getData().watched().inBackground().forPath(PATH);  
13.             System.out.println("修改後的data為" + new String(bs2 != null ? bs2 : newbyte[0]));  
14.             client.delete().forPath(PATH);  
15.             Stat stat = client.checkExists().forPath(PATH);  
16.             // Stat就是對zonde所有屬性的一個對映， stat=null表示節點不存在！
17.             System.out.println(stat);  
18.         } catch (Exception e) {  
19.             e.printStackTrace();  
20.         } finally {  
21.             CloseableUtils.closeQuietly(client);  
22.         }  
23.     }  
24. }  

 常用介面有

create()增

delete(): 刪

checkExists(): 判斷是否存在

setData():  改

getData(): 查

所有這些方法都以forpath()結尾，輔以watch(監聽)，withMode（指定模式），和inBackground（後臺執行）等方法來使用。

(補充) 1、建立連線

　　建立連線需要指定zk地址以及重試策略等，先上程式碼再解釋：

　　如註釋，建立客戶端連線我們通常需要指定重試策略，curator提供了3種重試機制，分別如上；對於fluent風格，就是每個操作都返回了一個物件，我們可以一直通過[.方法名]的方式書寫程式碼；client建立了之後，需要呼叫start方法才能真正去建立連線。會話超時時間是指當連線發生故障時，由於zk的心跳機制檢測，服務端認為會話超時的時間，會清除session；

　　2、建立、刪除、更新節點

　　連線建立之後，我們可以在伺服器上進行建立節點的操作，程式碼如下：

　　持久節點是persistent，我們建立的節點可能有好幾層，如果伺服器不存在父節點則會報錯並建立失敗，createingParentsIfNodeed()的作用是在父節點不存在的時候進行建立。刪除操作可能由於網路抖動等情況導致刪除失敗，由於節點資料操作一般對業務影響較大，故多數都會帶持續刪除的動作來確保正確刪除；節點更新刪除等操作若考慮版本校驗，則採用程式碼所示方式，在獲取節點資料的時候對節點狀態進行賦值，然後通過節點狀態可以獲得版本資訊。判斷節點是否存在，一般通過節點資訊判斷，若不存在，則節點資訊為null。

　　3、獲取位元組點列表

　　只有一行程式碼，返回string型別的list

　　4、非同步操作

　　非同步操作不會阻塞程式碼執行，對於操作完成後的業務處理，需要設定回撥函式來完成。以判斷節點是否存在為例：

　　非同步操作實際是在後臺另起一個執行緒來完成該操作，若執行緒較多勢必會影響伺服器效能，所以要用執行緒池來儘量降低對伺服器的消耗。需要考慮執行緒池的關閉操作，較繁瑣，不作贅述。

　　5、節點、子節點監聽

　　節點監聽需要用repices包中的NodeCache來完成，程式碼如下：

　　子節點的監聽需要用PathChildrenCache來完成，跟節點本身不一樣，程式碼如下：

　　6、許可權控制部分，略。

 此外，Curator還支援事務，一組crud操作同生同滅。程式碼如下

1. /** 
2.  * 事務操作 
3.  *  
4.  * @author shencl 
5.  */
6. publicclass TransactionExamples {  
7.     privatestatic CuratorFramework client = ClientFactory.newClient();  
8.     publicstaticvoid main(String[] args) {  
9.         try {  
10.             client.start();  
11.             // 開啟事務
12.             CuratorTransaction transaction = client.inTransaction();  
13.             Collection<CuratorTransactionResult> results = transaction.create()  
14.                     .forPath("/a/path", "some data".getBytes()).and().setData()  
15.                     .forPath("/another/path", "other data".getBytes()).and().delete().forPath("/yet/another/path")  
16.                     .and().commit();  
17.             for (CuratorTransactionResult result : results) {  
18.                 System.out.println(result.getForPath() + " - " + result.getType());  
19.             }  
20.         } catch (Exception e) {  
21.             e.printStackTrace();  
22.         } finally {  
23.             // 釋放客戶端連線
24.             CloseableUtils.closeQuietly(client);  
25.         }  
26.     }  
27. }  

 這段的程式碼的執行結果，由於最後一步delete的節點不存在，所以整個事務commit失敗。失敗的原因會放在Collection<CuratorTransactionResult>中，非常友好。

好了framework部分的內容就這麼多，是不是特別簡單呢。下面就來看看recipes包的內容吧。。

Recipes部分提供的功能官網列的很詳細，點選這裡。注意文章第一段：Curator宣稱，Recipes模組實現了除二階段提交之外的所有zookeeper特性。

二、Recipes模組

主要有

Elections(選舉)，Locks（鎖），Barriers（關卡），Atomic（原子量），Caches，Queues等

1、 Elections

選舉主要依賴於LeaderSelector和LeaderLatch2個類。前者是所有存活的客戶端不間斷的輪流做Leader，大同社會。後者是一旦選舉出Leader，除非有客戶端掛掉重新觸發選舉，否則不會交出領導權。某黨？

這兩者在實現上是可以切換的，直接上程式碼，怎麼切換註釋裡有。由於篇幅所限，這裡僅貼出基於LeaderSelector的選舉，更多程式碼見附件

1. /** 
2.  * 本類基於leaderSelector實現,所有存活的client會公平的輪流做leader 
3.  * 如果不想頻繁的變化Leader，需要在takeLeadership方法裡阻塞leader的變更！ 或者使用 {@link} 
4.  * LeaderLatchClient 
5.  */
6. publicclass LeaderSelectorClient extends LeaderSelectorListenerAdapter implements Closeable {  
7.     privatefinal String name;  
8.     privatefinal LeaderSelector leaderSelector;  
9.     privatefinal String PATH = "/leaderselector";  
10.     public LeaderSelectorClient(CuratorFramework client, String name) {  
11.         this.name = name;  
12.         leaderSelector = new LeaderSelector(client, PATH, this);  
13.         leaderSelector.autoRequeue();  
14.     }  
15.     publicvoid start() throws IOException {  
16.         leaderSelector.start();  
17.     }  
18.     @Override
19.     publicvoid close() throws IOException {  
20.         leaderSelector.close();  
21.     }  
22.     /** 
23.      * client成為leader後，會呼叫此方法 
24.      */
25.     @Override
26.     publicvoid takeLeadership(CuratorFramework client) throws Exception {  
27.         int waitSeconds = (int) (5 * Math.random()) + 1;  
28.         System.out.println(name + "是當前的leader");  
29.         try {  
30.             Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));  
31.         } catch (InterruptedException e) {  
32.             Thread.currentThread().interrupt();  
33.         } finally {  
34.             System.out.println(name + " 讓出領導權\n");  
35.         }  
36.     }  

1. /** 
2.  * leader選舉 
3.  *  
4.  * @author shencl 
5.  */
6. publicclass LeaderSelectorExample {  
7.     publicstaticvoid main(String[] args) {  
8.         List<CuratorFramework> clients = Lists.newArrayList();  
9.         List<LeaderSelectorClient> examples = Lists.newArrayList();  
10.         try {  
11.             for (int i = 0; i < 10; i++) {  
12.                 CuratorFramework client = ClientFactory.newClient();  
13.                 LeaderSelectorClient example = new LeaderSelectorClient(client, "Client #" + i);  
14.                 clients.add(client);  
15.                 examples.add(example);  
16.                 client.start();  
17.                 example.start();  
18.             }  
19.             System.out.println("----------先觀察一會選舉的結果-----------");  
20.             Thread.sleep(10000);  
21.             System.out.println("----------關閉前5個客戶端，再觀察選舉的結果-----------");  
22.             for (int i = 0; i < 5; i++) {  
23.                 clients.get(i).close();  
24.             }  
25.             // 這裡有個小技巧，讓main程式一直監聽控制檯輸入，非同步的程式碼就可以一直在執行。不同於while(ture)的是，按回車或esc可退出
26.             new BufferedReader(new InputStreamReader(System.in)).readLine();  
27.         } catch (Exception e) {  
28.             e.printStackTrace();  
29.         } finally {  
30.             for (LeaderSelectorClient exampleClient : examples) {  
31.                 CloseableUtils.closeQuietly(exampleClient);  
32.             }  
33.             for (CuratorFramework client : clients) {  
34.                 CloseableUtils.closeQuietly(client);  
35.             }  
36.         }  
37.     }  
38. }  

2、locks

curator lock相關的實現在recipes.locks包裡。頂級介面都是InterProcessLock。我們直接看最有代表性的InterProcessReadWriteLock 程序內部讀寫鎖（可重入讀寫鎖）。什麼叫可重入，什麼叫讀寫鎖。不清楚的先查好資料吧。總之讀寫鎖一定是成對出現的。    簡易傳送門

我們先定義兩個任務，可並行的執行的，和互斥執行的。

1. /** 
2.  * 並行任務 
3.  *  
4.  * @author shencl 
5.  */
6. publicclass ParallelJob implements Runnable {  
7.     privatefinal String name;  
8.     privatefinal InterProcessLock lock;  
9.     // 鎖等待時間
10.     privatefinalint wait_time = 5;  
11.     ParallelJob(String name, InterProcessLock lock) {  
12.         this.name = name;  
13.         this.lock = lock;  
14.     }  
15.     @Override
16.     publicvoid run() {  
17.         try {  
18.             doWork();  
19.         } catch (Exception e) {  
20.             // ingore;
21.         }  
22.     }  
23.     publicvoid doWork() throws Exception {  
24.         try {  
25.             if (!lock.acquire(wait_time, TimeUnit.SECONDS)) {  
26.                 System.err.println(name + "等待" + wait_time + "秒，仍未能獲取到lock,準備放棄。");  
27.             }  
28.             // 模擬job執行時間0-4000毫秒
29.             int exeTime = new Random().nextInt(4000);  
30.             System.out.println(name + "開始執行,預計執行時間= " + exeTime + "毫秒----------");  
31.             Thread.sleep(exeTime);  
32.         } catch (Exception e) {  
33.             e.printStackTrace();  
34.         } finally {  
35.             lock.release();  
36.         }  
37.     }  
38. }  

1. /** 
2.  * 互斥任務 
3.  *  
4.  * @author shencl 
5.  */
6. publicclass MutexJob implements Runnable {  
7.     privatefinal String name;  
8.     privatefinal InterProcessLock lock;  
9.     // 鎖等待時間
10.     privatefinalint wait_time = 10;  
11.     MutexJob(String name, InterProcessLock lock) {  
12.         this.name = name;  
13.         this.lock = lock;  
14.     }  
15.     @Override
16.     publicvoid run() {  
17.         try {  
18.             doWork();  
19.         } catch (Exception e) {  
20.             // ingore;
21.         }  
22.     }  
23.     publicvoid doWork() throws Exception {  
24.         try {  
25.             if (!lock.acquire(wait_time, TimeUnit.SECONDS)) {  
26.                 System.err.println(name + "等待" + wait_time + "秒，仍未能獲取到lock,準備放棄。");  
27.             }  
28.             // 模擬job執行時間0-2000毫秒
29.             int exeTime = new Random().nextInt(2000);  
30.             System.out.println(name + "開始執行,預計執行時間= " + exeTime + "毫秒----------");  
31.             Thread.sleep(exeTime);  
32.         } catch (Exception e) {  
33.             e.printStackTrace();  
34.         } finally {  
35.             lock.release();  
36.         }  
37.     }  
38. }  

鎖測試程式碼

1. /** 
2.  * 分散式鎖例項 
3.  *  
4.  * @author shencl 
5.  */
6. publicclass DistributedLockExample {  
7.     privatestatic CuratorFramework client = ClientFactory.newClient();  
8.     privatestaticfinal String PATH = "/locks";  
9.     // 程序內部（可重入）讀寫鎖
10.     privatestaticfinal InterProcessReadWriteLock lock;  
11.     // 讀鎖
12.     privatestaticfinal InterProcessLock readLock;  
13.     // 寫鎖
14.     privatestaticfinal InterProcessLock writeLock;  
15.     static {  
16.         client.start();  
17.         lock = new InterProcessReadWriteLock(client, PATH);  
18.         readLock = lock.readLock();  
19.         writeLock = lock.writeLock();  
20.     }  
21.     publicstaticvoid main(String[] args) {  
22.         try {  
23.             List<Thread> jobs = Lists.newArrayList();  
24.             for (int i = 0; i < 10; i++) {  
25.                 Thread t = new Thread(new ParallelJob("Parallel任務" + i, readLock));  
26.                 jobs.add(t);  
27.             }  
28.             for (int i = 0; i < 10; i++) {  
29.                 Thread t = new Thread(new MutexJob("Mutex任務" + i, writeLock));  
30.                 jobs.add(t);  
31.             }  
32.             for (Thread t : jobs) {  
33.                 t.start();  
34.             }  
35.         } catch (Exception e) {  
36.             e.printStackTrace();  
37.         } finally {  
38.             CloseableUtils.closeQuietly(client);  
39.         }  
40.     }  
41. }  

看到沒，用法和java concurrent包裡的ReentrantReadWriteLock 是一模一樣的。

事實上，整個recipes包的目錄結構、實現原理同java concurrent包的設定是很一致的。比如有queue，Semaphore，Barrier等類，。他整個就是模仿jdk的實現，只不過是基於分散式的！

後邊的幾項，Barriers（關卡），Atomic（原子量），Caches，Queues和java concurrent包裡的類的用法是一樣的，就不繼續貼了，有些附件裡有。

要說明的是：有的功能效能不是特別理想，網上也沒見有大的專案的使用案例。比如基於CAS機制的atomic，在某些情況重試的效率還不如硬同步，要是zookeeper節點再一多，各個節點之間通過event觸發的資料同步極其頻繁。那效能可以想象。

三、測試方法

 curator提供了很好的測試工具，你甚至是可以在完全沒有搭建zookeeper server端的情況下，完成測試。

有2個重要的類

TestingServer 模擬單點， TestingCluster模擬叢集。

需要使用的話，得依賴

1. <dependency>
2.     <groupId>org.apache.curator</groupId>
3.     <artifactId>curator-test</artifactId>
4.     <version>2.5.0</version>
5. </dependency>

全文完。

本文參考：

http://curator.apache.org/

http://www.cnblogs.com/hzhuxin/archive/2012/11/01/2749341.html

http://www.chengxuyuans.com/Java+/72042.html

http://macrochen.iteye.com/blog/1366136