Zookeeper開源客戶端框架Curator簡介

Curator是Netflix開源的一套ZooKeeper客戶端框架. Netflix在使用ZooKeeper的過程中發現ZooKeeper自帶的客戶端太底層, 應用方在使用的時候需要自己處理很多事情, 于是在它的基礎上包裝了一下, 提供了一套更好用的客戶端框架. Netflix在用ZooKeeper的過程中遇到的問題, 我們也遇到了, 所以開始研究一下, 首先從他在github上的源碼, wiki文檔以及Netflix的技術blog入手.?

看完官方的文檔之后, 發現Curator主要解決了三類問題:?

• 封裝ZooKeeper client與ZooKeeper server之間的連接處理;
• 提供了一套Fluent風格的操作API;
• 提供ZooKeeper各種應用場景(recipe, 比如共享鎖服務, 集群領導選舉機制)的抽象封裝.

Curator列舉的ZooKeeper使用過程中的幾個問題?
初始化連接的問題: 在client與server之間握手建立連接的過程中, 如果握手失敗, 執行所有的同步方法(比如create, getData等)將拋出異常?
自動恢復(failover)的問題: 當client與一臺server的連接丟失,并試圖去連接另外一臺server時, client將回到初始連接模式?
session過期的問題: 在極端情況下, 出現ZooKeeper session過期, 客戶端需要自己去監聽該狀態并重新創建ZooKeeper實例 .?
對可恢復異常的處理:當在server端創建一個有序ZNode, 而在將節點名返回給客戶端時崩潰, 此時client端拋出可恢復的異常, 用戶需要自己捕獲這些異常并進行重試?
使用場景的問題:Zookeeper提供了一些標準的使用場景支持, 但是ZooKeeper對這些功能的使用說明文檔很少, 而且很容易用錯. 在一些極端場景下如何處理, zk并沒有給出詳細的文檔說明. 比如共享鎖服務, 當服務器端創建臨時順序節點成功, 但是在客戶端接收到節點名之前掛掉了, 如果不能很好的處理這種情況, 將導致死鎖.?

Curator主要從以下幾個方面降低了zk使用的復雜性:?
重試機制:提供可插拔的重試機制, 它將給捕獲所有可恢復的異常配置一個重試策略, 并且內部也提供了幾種標準的重試策略(比如指數補償).?
連接狀態監控: Curator初始化之后會一直的對zk連接進行監聽, 一旦發現連接狀態發生變化, 將作出相應的處理.?
zk客戶端實例管理:Curator對zk客戶端到server集群連接進行管理. 并在需要的情況, 重建zk實例, 保證與zk集群的可靠連接?
各種使用場景支持:Curator實現zk支持的大部分使用場景支持(甚至包括zk自身不支持的場景), 這些實現都遵循了zk的最佳實踐, 并考慮了各種極端情況.?

Curator通過以上的處理, 讓用戶專注于自身的業務本身, 而無需花費更多的精力在zk本身.?

Curator聲稱的一些亮點:?

日志工具?
內部采用SLF4J 來輸出日志?
采用驅動器(driver)機制, 允許擴展和定制日志和跟蹤處理?
提供了一個TracerDriver接口, 通過實現addTrace()和addCount()接口來集成用戶自己的跟蹤框架?

和Curator相比, 另一個ZooKeeper客戶端——zkClient(https://github.com/sgroschupf/zkclient)的不足之處:?
文檔幾乎沒有?
異常處理弱爆了(簡單的拋出RuntimeException)?
重試處理太難用了?
沒有提供各種使用場景的實現?

對ZooKeeper自帶客戶端(ZooKeeper類)的"抱怨":?
只是一個底層實現?
要用需要自己寫大量的代碼?
很容易誤用?
需要自己處理連接丟失, 重試等?

Curator幾個組成部分?

• Client: 是ZooKeeper客戶端的一個替代品, 提供了一些底層處理和相關的工具方法.
• Framework: 用來簡化ZooKeeper高級功能的使用, 并增加了一些新的功能, 比如管理到ZooKeeper集群的連接, 重試處理
• Recipes: 實現了通用ZooKeeper的recipe, 該組件建立在Framework的基礎之上
• Utilities:各種ZooKeeper的工具類
• Errors: 異常處理, 連接, 恢復等.
• Extensions: recipe擴展

Client?
這是一個底層的API, 應用方基本對這個可以無視, 最好直接從Curator Framework入手?
主要包括三部分:?
不間斷連接管理?
連接重試處理?

Retry Loop(循環重試)?
一種典型的用法:?

RetryLoop?retryLoop?=?client.newRetryLoop();??
while?(?retryLoop.shouldContinue()?)??
{??
???try??
???{??
???????//?perform?your?work??
???????...??
???????//?it's?important?to?re-get?the?ZK?instance?as?there?may?have?been?an?error?and?the?instance?was?re-created??
???????ZooKeeper??????zk?=?client.getZookeeper();??
??
???????retryLoop.markComplete();??
???}??
???catch?(?Exception?e?)??
???{??
???????retryLoop.takeException(e);??
???}??
}??

如果在操作過程中失敗, 且這種失敗是可重試的, 而且在允許的次數內, Curator將保證操作的最終完成.?

另一種使用Callable接口的重試做法:?

RetryLoop.callWithRetry(client,?new?Callable()??
{??
??????@Override??
??????public?Void?call()?throws?Exception??
??????{??
??????????//?do?your?work?here?-?it?will?get?retried?if?needed??
??????????return?null;??
??????}??
});??

重試策略?
RetryPolicy接口只有一個方法(以前版本有兩個方法):?
public boolean allowRetry(int retryCount, long elapsedTimeMs);?
在開始重試之前, allowRetry方法被調用, 其參數將指定當前重試次數, 和操作已消耗時間. 如果允許, 將繼續重試, 否則拋出異常.?

Curator內部實現的幾種重試策略:?

• ExponentialBackoffRetry:重試指定的次數, 且每一次重試之間停頓的時間逐漸增加.
• RetryNTimes:指定最大重試次數的重試策略
• RetryOneTime:僅重試一次
• RetryUntilElapsed:一直重試直到達到規定的時間

Framework?
是ZooKeeper Client更高的抽象API?
自動連接管理: 當ZooKeeper客戶端內部出現異常, 將自動進行重連或重試, 該過程對外幾乎完全透明?
更清晰的API: 簡化了ZooKeeper原生的方法, 事件等, 提供流程的接口?

CuratorFrameworkFactory類提供了兩個方法, 一個工廠方法newClient, 一個構建方法build. 使用工廠方法newClient可以創建一個默認的實例, 而build構建方法可以對實例進行定制. 當CuratorFramework實例構建完成, 緊接著調用start()方法, 在應用結束的時候, 需要調用close()方法.? CuratorFramework是線程安全的. 在一個應用中可以共享同一個zk集群的CuratorFramework.?

CuratorFramework API采用了連貫風格的接口(Fluent Interface). 所有的操作一律返回構建器, 當所有元素加在一起之后, 整個方法看起來就像一個完整的句子. 比如下面的操作:?

client.create().forPath("/head",?new?byte[0]);??
client.delete().inBackground().forPath("/head");??
client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/head/child",?new?byte[0]);??
client.getData().watched().inBackground().forPath("/test");??

方法說明:?

• create(): 發起一個create操作. 可以組合其他方法 (比如mode 或background) 最后以forPath()方法結尾
• delete(): 發起一個刪除操作. 可以組合其他方法(version 或background) 最后以forPath()方法結尾
• checkExists(): 發起一個檢查ZNode 是否存在的操作. 可以組合其他方法(watch 或background) 最后以forPath()方法結尾
• getData(): 發起一個獲取ZNode數據的操作. 可以組合其他方法(watch, background 或get stat) 最后以forPath()方法結尾
• setData(): 發起一個設置ZNode數據的操作. 可以組合其他方法(version 或background) 最后以forPath()方法結尾
• getChildren(): 發起一個獲取ZNode子節點的操作. 可以組合其他方法(watch, background 或get stat) 最后以forPath()方法結尾
• inTransaction(): 發起一個ZooKeeper事務. 可以組合create, setData, check, 和/或delete 為一個操作, 然后commit() 提交

.?

通知(Notification)?
Curator的相關代碼已經更新了, 里面的接口已經由ClientListener改成CuratorListener了, 而且接口中去掉了clientCloseDueToError方法. 只有一個方法:?
eventReceived()??????????? 當一個后臺操作完成或者指定的watch被觸發時該方法被調用?

UnhandledErrorListener接口用來對異常進行處理.?

CuratorEvent(在以前版本為ClientEvent)是對各種操作觸發相關事件對象(POJO)的一個完整封裝, 而事件對象的內容跟事件類型相關, 下面是對應關系:?

CREATE	getResultCode() and getPath()
DELETE	getResultCode() and getPath()
EXISTS	getResultCode(), getPath() and getStat()
GET_DATA	getResultCode(), getPath(), getStat() and getData()
SET_DATA	getResultCode(), getPath() and getStat()
CHILDREN	getResultCode(), getPath(), getStat(), getChildren()
WATCHED	getWatchedEvent()

名稱空間(Namespace)?
因為一個zk集群會被多個應用共享, 為了避免各個應用的zk patch沖突, Curator Framework內部會給每一個Curator Framework實例分配一個namespace(可選). 這樣你在create ZNode的時候都會自動加上這個namespace作為這個node path的root. 使用代碼如下:?

CuratorFramework????client?=?CuratorFrameworkFactory.builder().namespace("MyApp")?...?build();??
?…??
client.create().forPath("/test",?data);??
//?node?was?actually?written?to:?"/MyApp/test"??

Recipe?

Curator實現ZooKeeper的所有recipe(除了兩段提交)?
選舉?
集群領導選舉(leader election)?

鎖服務?
共享鎖: 全局同步分布式鎖, 同一時間兩臺機器只有一臺能獲得同一把鎖.?
共享讀寫鎖: 用于分布式的讀寫互斥處理, 同時生成兩個鎖:一個讀鎖, 一個寫鎖, 讀鎖能被多個應用持有, 而寫鎖只能一個獨占, 當寫鎖未被持有時, 多個讀鎖持有者可以同時進行讀操作?
共享信號量: 在分布式系統中的各個JVM使用同一個zk lock path, 該path將跟一個給定數量的租約(lease)相關聯, 然后各個應用根據請求順序獲得對應的lease, 相對來說, 這是最公平的鎖服務使用方式.?
多共享鎖:內部構件多個共享鎖(會跟一個znode path關聯), 在acquire()過程中, 執行所有共享鎖的acquire()方法, 如果中間出現一個失敗, 則將釋放所有已require的共享鎖; 執行release()方法時, 則執行內部多個共享鎖的release方法(如果出現失敗將忽略)?

隊列(Queue)?
分布式隊列:采用持久順序zk node來實現FIFO隊列, 如果有多個消費者, 可以使用LeaderSelector來保證隊列的消費者順序?
分布式優先隊列: 優先隊列的分布式版本?
BlockingQueueConsumer: JDK阻塞隊列的分布式版本?

關卡(Barrier)?
分布式關卡:一堆客戶端去處理一堆任務, 只有所有的客戶端都執行完, 所有客戶端才能繼續往下處理?
雙分布式關卡:同時開始, 同時結束?

計數器(Counter)?
共享計數器:所有客戶端監聽同一個znode path, 并共享一個最新的integer計數值?
分布式AtomicLong(AtomicInteger): AtomicXxx的分布式版本, 先采用樂觀鎖更新, 若失敗再采用互斥鎖更新, 可以配置重試策略來處理重試?

工具類?

Path Cache?
Path Cache用于監聽ZNode的子節點的變化, 當add, update, remove子節點時將改變Path Cache state, 同時返回所有子節點的data和state.?
Curator中采用了PathChildrenCache類來處理Path Cache, 狀態的變化則采用PathChildrenCacheListener來監聽.
相關用法參見TestPathChildrenCache測試類?

注意: 當zk server的數據發生變化, zk client會出現不一致, 這個需要通過版本號來識別這種狀態的變化?

Test Server?
用來在測試中模擬一個本地進程內ZooKeeper Server.?

Test Cluster?
用來在測試中模擬一個ZooKeeper Server集群?

ZKPaths工具類?
提供了和ZNode相關的path處理工具方法:?
???

• getNodeFromPath: 根據給定path獲取node name. i.e. "/one/two/three" -> "three"
• ??? mkdirs: 根據給定路徑遞歸創建所有node
• ??? getSortedChildren: 根據給定路徑, 返回一個按序列號排序的子節點列表
• ??? makePath: 根據給定的path和子節點名, 創建一個完整path

EnsurePath工具類?

直接看例子, 具體的說就是調用多次, 只會執行一次創建節點操作.?

EnsurePath???????ensurePath?=?new?EnsurePath(aFullPathToEnsure);??
...??
String???????????nodePath?=?aFullPathToEnsure?+?"/foo";??
ensurePath.ensure(zk);???//?first?time?syncs?and?creates?if?needed??
zk.create(nodePath,?...);??
...??
ensurePath.ensure(zk);???//?subsequent?times?are?NOPs??
zk.create(nodePath,?...);??

Notification事件處理?
Curator對ZooKeeper的事件Watcher進行了封裝處理, 然后實現了一套監聽機制. 提供了幾個監聽接口用來處理ZooKeeper連接狀態的變化?
當連接出現異常, 將通過ConnectionStateListener接口進行監聽, 并進行相應的處理, 這些狀態變化包括:?

• 暫停(SUSPENDED): 當連接丟失, 將暫停所有操作, 直到連接重新建立, 如果在規定時間內無法建立連接, 將觸發LOST通知
• 重連(RECONNECTED): 連接丟失, 執行重連時, 將觸發該通知
• 丟失(LOST): 連接超時時, 將觸發該通知

從com.netflix.curator.framework.imps.CuratorFrameworkImpl.validateConnection(CuratorEvent)方法中我們可以知道, Curator分別將ZooKeeper的Disconnected, Expired, SyncConnected三種狀態轉換成上面三種狀態.?

參考?

• https://github.com/Netflix/curator
• https://github.com/sgroschupf/zkclient
• http://en.wikipedia.org/wiki/Fluent_interface
• http://huidian.iteye.com/blog/426664?fluent interface中文版
• http://techblog.netflix.com/2011/11/introducing-curator-netflix-zookeeper.html
• http://www.ibm.com/developerworks/cn/opensource/os-cn-zookeeper/?分布式服務框架 Zookeeper -- 管理分布式環境中的數據
• Netflix Curator 使用