簡介

Curator是Netflix公司開源的一套Zookeeper客戶端框架。瞭解過Zookeeper原生API都會清楚其複雜度。Curator幫助我們在其基礎上進行封裝、實現一些開發細節，包括接連重連、反覆註冊Watcher和NodeExistsException等。目前已經作為Apache的頂級專案出現，是最流行的Zookeeper客戶端之一。從編碼風格上來講，它提供了基於Fluent的程式設計風格支援。

除此之外，Curator還提供了Zookeeper的各種應用場景：Recipe、共享鎖服務、Master選舉機制和分散式計數器等。 

專案及依賴

版本

目前Curator有2.x.x和3.x.x兩個系列的版本，支援不同版本的Zookeeper。其中Curator 2.x.x相容Zookeeper的3.4.x和3.5.x。而Curator 3.x.x只相容Zookeeper 3.5.x，並且提供了一些諸如動態重新配置、watch刪除等新特性。

專案元件

Maven依賴

Curator的jar包已經發布到Maven中心，由以下幾個artifact的組成。根據需要選擇引入具體的artifact。但大多數情況下只用引入curator-recipes即可。

根據上面的描述，開發人員大多數情況下使用的都是curator-recipes的依賴，此依賴的maven配置如下：

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>2.12.0</version>
</dependency>

• 1
• 2
• 3
• 4
• 5

由於版本相容原因，採用了2.x.x的最高版本。

案例及功能說明

建立會話

Curator的建立會話方式與原生的API和ZkClient的建立方式區別很大。Curator建立客戶端是通過CuratorFrameworkFactory工廠類來實現的。其中，此工廠類提供了三種建立客戶端的方法。 
前兩種方法是通過newClient來實現，僅引數不同而已。

public static CuratorFramework newClient(String connectString, RetryPolicy retryPolicy)

public static CuratorFramework newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)

• 1
• 2
• 3

使用上面方法創建出一個CuratorFramework之後，需要再呼叫其start()方法完成會話建立。 
例項程式碼：

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
        CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181",retryPolicy);
        client.start();

• 1
• 2
• 3

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181",
                5000,1000,retryPolicy);
         client.start();

• 1
• 2
• 3
• 4

其中引數RetryPolicy提供重試策略的介面，可以讓使用者實現自定義的重試策略。預設提供了以下實現，分別為ExponentialBackoffRetry、BoundedExponentialBackoffRetry、RetryForever、RetryNTimes、RetryOneTime、RetryUntilElapsed。

進一步檢視原始碼可以得知，其實這兩種方法內部實現一樣，只是對外包裝成不同的方法。它們的底層都是通過第三個方法builder來實現的。 
例項程式碼：

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
        CuratorFramework client =CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .retryPolicy(retryPolicy)
                .sessionTimeoutMs(6000)
                .connectionTimeoutMs(3000)
                .build();
        client.start();

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8

觀察上面的例項，我們可以看到此處已經使用了Fluent風格的編碼。其中namespace(“demo”)這項設定用來定義此會話的獨立名稱空間，隨後的相應操作都是在此名稱空間下進行操作。

重試策略

上面的例子中使用到了ExponentialBackoffRetry重試策略實現。此策略先給定一個初始化sleep時間baseSleepTimeMs，在此基礎上結合重試次數，通過以下程式碼計算當前需要的sleep時間：

long sleepMs = baseSleepTimeMs * Math.max(1, random.nextInt(1 << (retryCount + 1)));
if ( sleepMs > maxSleepMs ){
            sleepMs = maxSleepMs;
 }

• 1
• 2
• 3
• 4

隨著重試次數的增加，計算出的sleep時間也會越來越大。如果超過maxSleepMs則使用maxSleepMs的時間。其中maxRetries限制了最大的嘗試次數。

建立節點

Curator建立節點的方法也是基於Fluent風格編碼，原生API中的引數很多都轉化為一層層的方法呼叫來進行設定。下面簡單介紹一下常用的幾個節點建立場景。 
（1）建立一個初始內容為空的節點

client.create().forPath(path);

• 1

Curator預設建立的是持久節點，內容為空。 
（2）建立一個包含內容的節點

client.create().forPath(path,"我是內容".getBytes());

• 1

Curator和ZkClient不同的是依舊採用Zookeeper原生API的風格，內容使用byte[]作為方法引數。 
（3）建立臨時節點，並遞迴建立父節點

client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);

• 1

此處Curator和ZkClient一樣封裝了遞迴建立父節點的方法。在遞迴建立父節點時，父節點為持久節點。

刪除節點

刪除節點的方法也是基於Fluent方式來進行操作，不同型別的操作呼叫新增不同的方法呼叫即可。 
（1）刪除一個子節點

client.delete().forPath(path);

• 1

（2）刪除節點並遞迴刪除其子節點

client.delete().deletingChildrenIfNeeded().forPath(path);

• 1

（3）指定版本進行刪除

client.delete().withVersion(1).forPath(path);

• 1

如果此版本已經不存在，則刪除異常，異常資訊如下。

org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion for

• 1

（4）強制保證刪除一個節點

client.delete().guaranteed().forPath(path);

• 1

只要客戶端會話有效，那麼Curator會在後臺持續進行刪除操作，直到節點刪除成功。比如遇到一些網路異常的情況，此guaranteed的強制刪除就會很有效果。

讀取資料

讀取節點資料內容API相當簡單，Curator提供了傳入一個Stat，使用節點當前的Stat替換到傳入的Stat的方法，查詢方法執行完成之後，Stat引用已經執行當前最新的節點Stat。

// 普通查詢
client.getData().forPath(path);
// 包含狀態查詢
Stat stat = new Stat();
client.getData().storingStatIn(stat()).forPath(path);

• 1
• 2
• 3
• 4
• 5

更新資料

更新資料，如果未傳入version引數，那麼更新當前最新版本，如果傳入version則更新指定version，如果version已經變更，則丟擲異常。

// 普通更新
client.setData().forPath(path,"新內容".getBytes());
// 指定版本更新
client.setData().withVersion(1).forPath(path);

• 1
• 2
• 3
• 4

版本不一直異常資訊：

org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion for

• 1

非同步介面

在使用以上針對節點的操作API時，我們會發現每個介面都有一個inBackground()方法可供呼叫。此介面就是Curator提供的非同步呼叫入口。對應的非同步處理介面為BackgroundCallback。此介面指提供了一個processResult的方法，用來處理回撥結果。其中processResult的引數event中的getType()包含了各種事件型別，getResultCode()包含了各種響應碼。

重點說一下inBackground的以下介面：

public T inBackground(BackgroundCallback callback, Executor executor);

• 1

此介面就允許傳入一個Executor例項，用一個專門執行緒池來處理返回結果之後的業務邏輯。