zookeepercurator
Curator是Netflix開源的一套ZooKeeper客戶端框架. Netflix在使用ZooKeeper的過程中發(fā)現(xiàn)ZooKeeper自帶的客戶端太底層, 應(yīng)用方在使用的時候需要自己處理很多事情, 于是在它的基礎(chǔ)上包裝了一下, 提供了一套更好用的客戶端框架. Netflix在用ZooKeeper的過程中遇到的問題, 我們也遇到了, 所以開始研究一下, 首先從他在github上的源碼, wiki文檔以及Netflix的技術(shù)blog入手.
看完官方的文檔之后, 發(fā)現(xiàn)Curator主要解決了三類問題:
1.封裝ZooKeeper client與ZooKeeper server之間的連接處理;
2.提供了一套Fluent風(fēng)格的操作API;
3.提供ZooKeeper各種應(yīng)用場景(recipe, 比如共享鎖服務(wù), 集群領(lǐng)導(dǎo)選舉機(jī)制)的抽象封裝.
Curator列舉的ZooKeeper使用過程中的幾個問題
1.初始化連接的問題: 在client與server之間握手建立連接的過程中, 如果握手失敗, 執(zhí)行所有的同步方法(比如create, getData等)將拋出異常
2.自動恢復(fù)(failover)的問題: 當(dāng)client與一臺server的連接丟失,并試圖去連接另外一臺server時, client將回到初始連接模式
session過期的問題: 在極端情況下, 出現(xiàn)ZooKeeper session過期, 客戶端需要自己去監(jiān)聽該狀態(tài)并重新創(chuàng)建ZooKeeper實例 .
3.對可恢復(fù)異常的處理:當(dāng)在server端創(chuàng)建一個有序ZNode, 而在將節(jié)點名返回給客戶端時崩潰, 此時client端拋出可恢復(fù)的異常, 用戶需要自己捕獲這些異常并進(jìn)行重試
4.使用場景的問題:Zookeeper提供了一些標(biāo)準(zhǔn)的使用場景支持, 但是ZooKeeper對這些功能的使用說明文檔很少, 而且很容易用錯. 在一些極端場景下如何處理, zk并沒有給出詳細(xì)的文檔說明. 比如共享鎖服務(wù), 當(dāng)服務(wù)器端創(chuàng)建臨時順序節(jié)點成功, 但是在客戶端接收到節(jié)點名之前掛掉了, 如果不能很好的處理這種情況, 將導(dǎo)致死鎖.
Curator主要從以下幾個方面降低了zk使用的復(fù)雜性:
1.重試機(jī)制:提供可插拔的重試機(jī)制, 它將給捕獲所有可恢復(fù)的異常配置一個重試策略, 并且內(nèi)部也提供了幾種標(biāo)準(zhǔn)的重試策略(比如指數(shù)補(bǔ)償).
2.連接狀態(tài)監(jiān)控: Curator初始化之后會一直的對zk連接進(jìn)行監(jiān)聽, 一旦發(fā)現(xiàn)連接狀態(tài)發(fā)生變化, 將作出相應(yīng)的處理.
3.zk客戶端實例管理:Curator對zk客戶端到server集群連接進(jìn)行管理. 并在需要的情況, 重建zk實例, 保證與zk集群的可靠連接
各種使用場景支持:Curator實現(xiàn)zk支持的大部分使用場景支持(甚至包括zk自身不支持的場景), 這些實現(xiàn)都遵循了zk的較佳實踐, 并考慮了各種極端情況.
Curator通過以上的處理, 讓用戶專注于自身的業(yè)務(wù)本身, 而無需花費更多的精力在zk本身.
Curator聲稱的一些亮點:
日志工具
內(nèi)部采用SLF4J 來輸出日志
采用驅(qū)動器(driver)機(jī)制, 允許擴(kuò)展和定制日志和跟蹤處理
提供了一個TracerDriver接口, 通過實現(xiàn)addTrace()和addCount()接口來集成用戶自己的跟蹤框架
和Curator相比, 另一個ZooKeeper客戶端——zkClient
文檔幾乎沒有
異常處理弱爆了(簡單的拋出RuntimeException)
重試處理太難用了
沒有提供各種使用場景的實現(xiàn)
對ZooKeeper自帶客戶端(ZooKeeper類)的"抱怨":
只是一個底層實現(xiàn)
要用需要自己寫大量的代碼
很容易誤用
需要自己處理連接丟失, 重試等
Curator幾個組成部分
1.Client: 是ZooKeeper客戶端的一個替代品, 提供了一些底層處理和相關(guān)的工具方法.
2.Framework: 用來簡化ZooKeeper高級功能的使用, 并增加了一些新的功能, 比如管理到ZooKeeper集群的連接, 重試處理
3.Recipes: 實現(xiàn)了通用ZooKeeper的recipe, 該組件建立在Framework的基礎(chǔ)之上
4.Utilities:各種ZooKeeper的工具類
5.Errors: 異常處理, 連接, 恢復(fù)等.
6.Extensions: recipe擴(kuò)展
Client
這是一個底層的API, 應(yīng)用方基本對這個可以無視, 較好直接從Curator Framework入手，主要包括三部分:
不間斷連接管理
連接重試處理
Retry Loop(循環(huán)重試)
一種典型的用法:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
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); }} |
如果在操作過程中失敗, 且這種失敗是可重試的, 而且在允許的次數(shù)內(nèi), Curator將保證操作的最終完成.
另一種使用Callable接口的重試做法:
|
1
2
3
4
5
6
7
8
9
|
RetryLoop.callWithRetry(client, new Callable(){ @Override public Void call() throws Exception { // do your work here - it will get retried if needed return null; }}); |
重試策略
RetryPolicy接口只有一個方法(以前版本有兩個方法):
|
1
|
public boolean allowRetry(int retryCount, long elapsedTimeMs); |
在開始重試之前, allowRetry方法被調(diào)用, 其參數(shù)將指定當(dāng)前重試次數(shù), 和操作已消耗時間. 如果允許, 將繼續(xù)重試, 否則拋出異常.
Curator內(nèi)部實現(xiàn)的幾種重試策略:
1.ExponentialBackoffRetry:重試指定的次數(shù), 且每一次重試之間停頓的時間逐漸增加.
2.RetryNTimes:指定較大重試次數(shù)的重試策略
3.RetryOneTime:僅重試一次
4.RetryUntilElapsed:一直重試直到達(dá)到規(guī)定的時間
Framework
是ZooKeeper Client更高的抽象API
自動連接管理: 當(dāng)ZooKeeper客戶端內(nèi)部出現(xiàn)異常, 將自動進(jìn)行重連或重試, 該過程對外幾乎完全透明
更清晰的API: 簡化了ZooKeeper原生的方法, 事件等, 提供流程的接口
CuratorFrameworkFactory類提供了兩個方法, 一個工廠方法newClient, 一個構(gòu)建方法build. 使用工廠方法newClient可以創(chuàng)建一個默認(rèn)的實例, 而build構(gòu)建方法可以對實例進(jìn)行定制. 當(dāng)CuratorFramework實例構(gòu)建完成, 緊接著調(diào)用start()方法, 在應(yīng)用結(jié)束的時候, 需要調(diào)用close()方法. CuratorFramework是線程安全的. 在一個應(yīng)用中可以共享同一個zk集群的CuratorFramework.
CuratorFramework API采用了連貫風(fēng)格的接口(Fluent Interface). 所有的操作一律返回構(gòu)建器, 當(dāng)所有元素加在一起之后, 整個方法看起來就像一個完整的句子. 比如下面的操作:
|
1
2
3
4
|
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"); |
方法說明:
1.create(): 發(fā)起一個create操作. 可以組合其他方法 (比如mode 或background) 最后以forPath()方法結(jié)尾
2.delete(): 發(fā)起一個刪除操作. 可以組合其他方法(version 或background) 最后以forPath()方法結(jié)尾
3.checkExists(): 發(fā)起一個檢查ZNode 是否存在的操作. 可以組合其他方法(watch 或background) 最后以forPath()方法結(jié)尾
4.getData(): 發(fā)起一個獲取ZNode數(shù)據(jù)的操作. 可以組合其他方法(watch, background 或get stat) 最后以forPath()方法結(jié)尾
5.setData(): 發(fā)起一個設(shè)置ZNode數(shù)據(jù)的操作. 可以組合其他方法(version 或background) 最后以forPath()方法結(jié)尾
6.getChildren(): 發(fā)起一個獲取ZNode子節(jié)點的操作. 可以組合其他方法(watch, background 或get stat) 最后以forPath()方法結(jié)尾
7.inTransaction(): 發(fā)起一個ZooKeeper事務(wù). 可以組合create, setData, check, 和/或delete 為一個操作, 然后commit() 提交
.
通知(Notification)
Curator的相關(guān)代碼已經(jīng)更新了, 里面的接口已經(jīng)由ClientListener改成CuratorListener了, 而且接口中去掉了clientCloseDueToError方法. 只有一個方法:
eventReceived() 當(dāng)一個后臺操作完成或者指定的watch被觸發(fā)時該方法被調(diào)用
UnhandledErrorListener接口用來對異常進(jìn)行處理.
CuratorEvent(在以前版本為ClientEvent)是對各種操作觸發(fā)相關(guān)事件對象(POJO)的一個完整封裝, 而事件對象的內(nèi)容跟事件類型相關(guān), 下面是對應(yīng)關(guān)系:
| 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集群會被多個應(yīng)用共享, 為了避免各個應(yīng)用的zk patch沖突, Curator Framework內(nèi)部會給每一個Curator Framework實例分配一個namespace(可選). 這樣你在create ZNode的時候都會自動加上這個namespace作為這個node path的root. 使用代碼如下:
|
1
2
3
4
|
CuratorFramework client = CuratorFrameworkFactory.builder().namespace("MyApp") ... build();…client.create().forPath("/test", data);// node was actually written to: "/MyApp/test" |
Recipe
Curator實現(xiàn)ZooKeeper的所有recipe(除了兩段提交)
選舉
集群領(lǐng)導(dǎo)選舉(leader election)
鎖服務(wù)
共享鎖: 全局同步分布式鎖, 同一時間兩臺機(jī)器只有一臺能獲得同一把鎖.
共享讀寫鎖: 用于分布式的讀寫互斥處理, 同時生成兩個鎖:一個讀鎖, 一個寫鎖, 讀鎖能被多個應(yīng)用持有, 而寫鎖只能一個獨占, 當(dāng)寫鎖未被持有時, 多個讀鎖持有者可以同時進(jìn)行讀操作
共享信號量: 在分布式系統(tǒng)中的各個JVM使用同一個zk lock path, 該path將跟一個給定數(shù)量的租約(lease)相關(guān)聯(lián), 然后各個應(yīng)用根據(jù)請求順序獲得對應(yīng)的lease, 相對來說, 這是最公平的鎖服務(wù)使用方式.
多共享鎖:內(nèi)部構(gòu)件多個共享鎖(會跟一個znode path關(guān)聯(lián)), 在acquire()過程中, 執(zhí)行所有共享鎖的acquire()方法, 如果中間出現(xiàn)一個失敗, 則將釋放所有已require的共享鎖; 執(zhí)行release()方法時, 則執(zhí)行內(nèi)部多個共享鎖的release方法(如果出現(xiàn)失敗將忽略)
隊列(Queue)
分布式隊列:采用持久順序zk node來實現(xiàn)FIFO隊列, 如果有多個消費者, 可以使用LeaderSelector來保證隊列的消費者順序
分布式優(yōu)先隊列: 優(yōu)先隊列的分布式版本
BlockingQueueConsumer: JDK阻塞隊列的分布式版本
關(guān)卡(Barrier)
分布式關(guān)卡:一堆客戶端去處理一堆任務(wù), 只有所有的客戶端都執(zhí)行完, 所有客戶端才能繼續(xù)往下處理
雙分布式關(guān)卡:同時開始, 同時結(jié)束
計數(shù)器(Counter)
共享計數(shù)器:所有客戶端監(jiān)聽同一個znode path, 并共享一個的integer計數(shù)值
分布式AtomicLong(AtomicInteger): AtomicXxx的分布式版本, 先采用樂觀鎖更新, 若失敗再采用互斥鎖更新, 可以配置重試策略來處理重試
工具類
Path Cache
Path Cache用于監(jiān)聽ZNode的子節(jié)點的變化, 當(dāng)add, update, remove子節(jié)點時將改變Path Cache state, 同時返回所有子節(jié)點的data和state.
Curator中采用了PathChildrenCache類來處理Path Cache, 狀態(tài)的變化則采用PathChildrenCacheListener來監(jiān)聽.
相關(guān)用法參見TestPathChildrenCache測試類
注意: 當(dāng)zk server的數(shù)據(jù)發(fā)生變化, zk client會出現(xiàn)不一致, 這個需要通過版本號來識別這種狀態(tài)的變化
Test Server
用來在測試中模擬一個本地進(jìn)程內(nèi)ZooKeeper Server.
Test Cluster
用來在測試中模擬一個ZooKeeper Server集群
ZKPaths工具類
提供了和ZNode相關(guān)的path處理工具方法:
1.getNodeFromPath: 根據(jù)給定path獲取node name. i.e. "/one/two/three" -> "three"
2.mkdirs: 根據(jù)給定路徑遞歸創(chuàng)建所有node
3.getSortedChildren: 根據(jù)給定路徑, 返回一個按序列號排序的子節(jié)點列表
4.makePath: 根據(jù)給定的path和子節(jié)點名, 創(chuàng)建一個完整path
EnsurePath工具類
直接看例子, 具體的說就是調(diào)用多次, 只會執(zhí)行一次創(chuàng)建節(jié)點操作.
|
1
2
3
4
5
6
7
8
|
EnsurePath ensurePath = new EnsurePath(aFullPathToEnsure);...String nodePath = aFullPathToEnsure + "/foo";ensurePath.ensure(zk); // first time syncs and creates if neededzk.create(nodePath, ...);...ensurePath.ensure(zk); // subsequent times are NOPszk.create(nodePath, ...); |
Notification事件處理
Curator對ZooKeeper的事件Watcher進(jìn)行了封裝處理, 然后實現(xiàn)了一套監(jiān)聽機(jī)制. 提供了幾個監(jiān)聽接口用來處理ZooKeeper連接狀態(tài)的變化
當(dāng)連接出現(xiàn)異常, 將通過ConnectionStateListener接口進(jìn)行監(jiān)聽, 并進(jìn)行相應(yīng)的處理, 這些狀態(tài)變化包括:
1.暫停(SUSPENDED): 當(dāng)連接丟失, 將暫停所有操作, 直到連接重新建立, 如果在規(guī)定時間內(nèi)無法建立連接, 將觸發(fā)LOST通知
2.重連(RECONNECTED): 連接丟失, 執(zhí)行重連時, 將觸發(fā)該通知
3.丟失(LOST): 連接超時時, 將觸發(fā)該通知
從com.netflix.curator.framework.imps.CuratorFrameworkImpl.validateConnection(CuratorEvent)方法中我們可以知道, Curator分別將ZooKeeper的Disconnected, Expired, SyncConnected三種狀態(tài)轉(zhuǎn)換成上面三種狀態(tài).
總結(jié)
以上就是本文關(guān)于淺談Zookeeper開源客戶端框架Curator的全部內(nèi)容，希望對大家有所幫助。如有不足之處，歡迎留言指正。感謝朋友們對本站的支持！
原文鏈接：http://f.dataguru.cn/thread-120125-1-1.html
