ZK客戶端Curator使用詳解
zookeeper不是為高可用性設(shè)計的,但它使用ZAB協(xié)議達(dá)到了極高的一致性,所以是個CP系統(tǒng)。所以它經(jīng)常被選作注冊中心、配置中心、分布式鎖等場景。
它的性能是非常有限的,而且API并不是那么好用。xjjdog傾向于使用基于Raft協(xié)議的Etcd或者Consul,它們更加輕量級一些。
Curator是netflix公司開源的一套zookeeper客戶端,目前是Apache的頂級項目。與Zookeeper提供的原生客戶端相比,Curator的抽象層次更高,簡化了Zookeeper客戶端的開發(fā)量。Curator解決了很多zookeeper客戶端非常底層的細(xì)節(jié)開發(fā)工作,包括連接重連、反復(fù)注冊wathcer和NodeExistsException 異常等。
Curator由一系列的模塊構(gòu)成,對于一般開發(fā)者而言,常用的是curator-framework和curator-recipes,下面對此依次介紹。
1.maven依賴
最新版本的curator 4.3.0支持zookeeper 3.4.x和3.5,但是需要注意curator傳遞進來的依賴,需要和實際服務(wù)器端使用的版本相符,以我們目前使用的zookeeper 3.4.6為例。
- <dependency>
 - <groupId>org.apache.curator</groupId>
 - <artifactId>curator-framework</artifactId>
 - <version>4.3.0</version>
 - <exclusions>
 - <exclusion>
 - <groupId>org.apache.zookeeper</groupId>
 - <artifactId>zookeeper</artifactId>
 - </exclusion>
 - </exclusions>
 - </dependency>
 - <dependency>
 - <groupId>org.apache.curator</groupId>
 - <artifactId>curator-recipes</artifactId>
 - <version>4.3.0</version>
 - <exclusions>
 - <exclusion>
 - <groupId>org.apache.zookeeper</groupId>
 - <artifactId>zookeeper</artifactId>
 - </exclusion>
 - </exclusions>
 - </dependency>
 - <dependency>
 - <groupId>org.apache.zookeeper</groupId>
 - <artifactId>zookeeper</artifactId>
 - <version>3.4.6</version>
 - </dependency>
 
2.curator-framework
下面是一些常見的zk相關(guān)的操作。
- public static CuratorFramework getClient() {
 - return CuratorFrameworkFactory.builder()
 - .connectString("127.0.0.1:2181")
 - .retryPolicy(new ExponentialBackoffRetry(1000, 3))
 - .connectionTimeoutMs(15 * 1000) //連接超時時間,默認(rèn)15秒
 - .sessionTimeoutMs(60 * 1000) //會話超時時間,默認(rèn)60秒
 - .namespace("arch") //設(shè)置命名空間
 - .build();
 - }
 - public static void create(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
 - client.create().creatingParentsIfNeeded().forPath(path, payload);
 - }
 - public static void createEphemeral(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
 - client.create().withMode(CreateMode.EPHEMERAL).forPath(path, payload);
 - }
 - public static String createEphemeralSequential(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
 - return client.create().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, payload);
 - }
 - public static void setData(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
 - client.setData().forPath(path, payload);
 - }
 - public static void delete(final CuratorFramework client, final String path) throws Exception {
 - client.delete().deletingChildrenIfNeeded().forPath(path);
 - }
 - public static void guaranteedDelete(final CuratorFramework client, final String path) throws Exception {
 - client.delete().guaranteed().forPath(path);
 - }
 - public static String getData(final CuratorFramework client, final String path) throws Exception {
 - return new String(client.getData().forPath(path));
 - }
 - public static List<String> getChildren(final CuratorFramework client, final String path) throws Exception {
 - return client.getChildren().forPath(path);
 - }
 
3.curator-recipescurator-recipes
提供了一些zk的典型使用場景的參考。下面主要介紹一下開發(fā)中常用的組件。
事件監(jiān)聽
zookeeper原生支持通過注冊watcher來進行事件監(jiān)聽,但是其使用不是特別方便,需要開發(fā)人員自己反復(fù)注冊watcher,比較繁瑣。
Curator引入Cache來實現(xiàn)對zookeeper服務(wù)端事務(wù)的監(jiān)聽。Cache是Curator中對事件監(jiān)聽的包裝,其對事件的監(jiān)聽其實可以近似看作是一個本地緩存視圖和遠(yuǎn)程Zookeeper視圖的對比過程。同時,Curator能夠自動為開發(fā)人員處理反復(fù)注冊監(jiān)聽,從而大大簡化原生api開發(fā)的繁瑣過程。
1)Node Cache
- public static void nodeCache() throws Exception {
 - final String path = "/nodeCache";
 - final CuratorFramework client = getClient();
 - client.start();
 - delete(client, path);
 - create(client, path, "cache".getBytes());
 - final NodeCache nodeCache = new NodeCache(client, path);
 - nodeCache.start(true);
 - nodeCache.getListenable()
 - .addListener(() -> System.out.println("node data change, new data is " + new String(nodeCache.getCurrentData().getData())));
 - setData(client, path, "cache1".getBytes());
 - setData(client, path, "cache2".getBytes());
 - Thread.sleep(1000);
 - client.close();
 - }
 
NodeCache可以監(jiān)聽指定的節(jié)點,注冊監(jiān)聽器后,節(jié)點的變化會通知相應(yīng)的監(jiān)聽器
2)Path Cache
Path Cache 用來監(jiān)聽ZNode的子節(jié)點事件,包括added、updateed、removed,Path Cache會同步子節(jié)點的狀態(tài),產(chǎn)生的事件會傳遞給注冊的PathChildrenCacheListener。
- public static void pathChildrenCache() throws Exception {
 - final String path = "/pathChildrenCache";
 - final CuratorFramework client = getClient();
 - client.start();
 - final PathChildrenCache cache = new PathChildrenCache(client, path, true);
 - cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
 - cache.getListenable().addListener((client1, event) -> {
 - switch (event.getType()) {
 - case CHILD_ADDED:
 - System.out.println("CHILD_ADDED:" + event.getData().getPath());
 - break;
 - case CHILD_REMOVED:
 - System.out.println("CHILD_REMOVED:" + event.getData().getPath());
 - break;
 - case CHILD_UPDATED:
 - System.out.println("CHILD_UPDATED:" + event.getData().getPath());
 - break;
 - case CONNECTION_LOST:
 - System.out.println("CONNECTION_LOST:" + event.getData().getPath());
 - break;
 - case CONNECTION_RECONNECTED:
 - System.out.println("CONNECTION_RECONNECTED:" + event.getData().getPath());
 - break;
 - case CONNECTION_SUSPENDED:
 - System.out.println("CONNECTION_SUSPENDED:" + event.getData().getPath());
 - break;
 - case INITIALIZED:
 - System.out.println("INITIALIZED:" + event.getData().getPath());
 - break;
 - default:
 - break;
 - }
 - });
 - // client.create().withMode(CreateMode.PERSISTENT).forPath(path);
 - Thread.sleep(1000);
 - client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
 - Thread.sleep(1000);
 - client.delete().forPath(path + "/c1");
 - Thread.sleep(1000);
 - client.delete().forPath(path); //監(jiān)聽節(jié)點本身的變化不會通知
 - Thread.sleep(1000);
 - client.close();
 - }
 
3)Tree Cache
Path Cache和Node Cache的“合體”,監(jiān)視路徑下的創(chuàng)建、更新、刪除事件,并緩存路徑下所有孩子結(jié)點的數(shù)據(jù)。
- public static void treeCache() throws Exception {
 - final String path = "/treeChildrenCache";
 - final CuratorFramework client = getClient();
 - client.start();
 - final TreeCache cache = new TreeCache(client, path);
 - cache.start();
 - cache.getListenable().addListener((client1, event) -> {
 - switch (event.getType()){
 - case NODE_ADDED:
 - System.out.println("NODE_ADDED:" + event.getData().getPath());
 - break;
 - case NODE_REMOVED:
 - System.out.println("NODE_REMOVED:" + event.getData().getPath());
 - break;
 - case NODE_UPDATED:
 - System.out.println("NODE_UPDATED:" + event.getData().getPath());
 - break;
 - case CONNECTION_LOST:
 - System.out.println("CONNECTION_LOST:" + event.getData().getPath());
 - break;
 - case CONNECTION_RECONNECTED:
 - System.out.println("CONNECTION_RECONNECTED:" + event.getData().getPath());
 - break;
 - case CONNECTION_SUSPENDED:
 - System.out.println("CONNECTION_SUSPENDED:" + event.getData().getPath());
 - break;
 - case INITIALIZED:
 - System.out.println("INITIALIZED:" + event.getData().getPath());
 - break;
 - default:
 - break;
 - }
 - });
 - client.create().withMode(CreateMode.PERSISTENT).forPath(path);
 - Thread.sleep(1000);
 - client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
 - Thread.sleep(1000);
 - setData(client, path, "test".getBytes());
 - Thread.sleep(1000);
 - client.delete().forPath(path + "/c1");
 - Thread.sleep(1000);
 - client.delete().forPath(path);
 - Thread.sleep(1000);
 - client.close();
 - }
 
選舉
curator提供了兩種方式,分別是Leader Latch和Leader Election。
1)Leader Latch
隨機從候選著中選出一臺作為leader,選中之后除非調(diào)用close()釋放leadship,否則其他的后選擇無法成為leader
- public class LeaderLatchTest {
 - private static final String PATH = "/demo/leader";
 - public static void main(String[] args) {
 - List<LeaderLatch> latchList = new ArrayList<>();
 - List<CuratorFramework> clients = new ArrayList<>();
 - try {
 - for (int i = 0; i < 10; i++) {
 - CuratorFramework client = getClient();
 - client.start();
 - clients.add(client);
 - final LeaderLatch leaderLatch = new LeaderLatch(client, PATH, "client#" + i);
 - leaderLatch.addListener(new LeaderLatchListener() {
 - @Override
 - public void isLeader() {
 - System.out.println(leaderLatch.getId() + ":I am leader. I am doing jobs!");
 - }
 - @Override
 - public void notLeader() {
 - System.out.println(leaderLatch.getId() + ":I am not leader. I will do nothing!");
 - }
 - });
 - latchList.add(leaderLatch);
 - leaderLatch.start();
 - }
 - Thread.sleep(1000 * 60);
 - } catch (Exception e) {
 - e.printStackTrace();
 - } finally {
 - for (CuratorFramework client : clients) {
 - CloseableUtils.closeQuietly(client);
 - }
 - for (LeaderLatch leaderLatch : latchList) {
 - CloseableUtils.closeQuietly(leaderLatch);
 - }
 - }
 - }
 - public static CuratorFramework getClient() {
 - return CuratorFrameworkFactory.builder()
 - .connectString("127.0.0.1:2181")
 - .retryPolicy(new ExponentialBackoffRetry(1000, 3))
 - .connectionTimeoutMs(15 * 1000) //連接超時時間,默認(rèn)15秒
 - .sessionTimeoutMs(60 * 1000) //會話超時時間,默認(rèn)60秒
 - .namespace("arch") //設(shè)置命名空間
 - .build();
 - }
 - }
 
2)Leader Election
通過LeaderSelectorListener可以對領(lǐng)導(dǎo)權(quán)進行控制, 在適當(dāng)?shù)臅r候釋放領(lǐng)導(dǎo)權(quán),這樣每個節(jié)點都有可能獲得領(lǐng)導(dǎo)權(quán)。而LeaderLatch則一直持有l(wèi)eadership, 除非調(diào)用close方法,否則它不會釋放領(lǐng)導(dǎo)權(quán)。
- public class LeaderSelectorTest {
 - private static final String PATH = "/demo/leader";
 - public static void main(String[] args) {
 - List<LeaderSelector> selectors = new ArrayList<>();
 - List<CuratorFramework> clients = new ArrayList<>();
 - try {
 - for (int i = 0; i < 10; i++) {
 - CuratorFramework client = getClient();
 - client.start();
 - clients.add(client);
 - final String name = "client#" + i;
 - LeaderSelector leaderSelector = new LeaderSelector(client, PATH, new LeaderSelectorListenerAdapter() {
 - @Override
 - public void takeLeadership(CuratorFramework client) throws Exception {
 - System.out.println(name + ":I am leader.");
 - Thread.sleep(2000);
 - }
 - });
 - leaderSelector.autoRequeue();
 - leaderSelector.start();
 - selectors.add(leaderSelector);
 - }
 - Thread.sleep(Integer.MAX_VALUE);
 - } catch (Exception e) {
 - e.printStackTrace();
 - } finally {
 - for (CuratorFramework client : clients) {
 - CloseableUtils.closeQuietly(client);
 - }
 - for (LeaderSelector selector : selectors) {
 - CloseableUtils.closeQuietly(selector);
 - }
 - }
 - }
 - public static CuratorFramework getClient() {
 - return CuratorFrameworkFactory.builder()
 - .connectString("127.0.0.1:2181")
 - .retryPolicy(new ExponentialBackoffRetry(1000, 3))
 - .connectionTimeoutMs(15 * 1000) //連接超時時間,默認(rèn)15秒
 - .sessionTimeoutMs(60 * 1000) //會話超時時間,默認(rèn)60秒
 - .namespace("arch") //設(shè)置命名空間
 - .build();
 - }
 - }
 
分布式鎖
1)可重入鎖Shared Reentrant Lock
Shared意味著鎖是全局可見的, 客戶端都可以請求鎖。Reentrant和JDK的ReentrantLock類似, 意味著同一個客戶端在擁有鎖的同時,可以多次獲取,不會被阻塞。它是由類InterProcessMutex來實現(xiàn)。它的構(gòu)造函數(shù)為:
- public InterProcessMutex(CuratorFramework client, String path)
 
通過acquire獲得鎖,并提供超時機制:
- /**
 - * Acquire the mutex - blocking until it's available. Note: the same thread can call acquire
 - * re-entrantly. Each call to acquire must be balanced by a call to release()
 - */
 - public void acquire();
 - /**
 - * Acquire the mutex - blocks until it's available or the given time expires. Note: the same thread can
 - * call acquire re-entrantly. Each call to acquire that returns true must be balanced by a call to release()
 - * Parameters:
 - * time - time to wait
 - * unit - time unit
 - * Returns:
 - * true if the mutex was acquired, false if not
 - */
 - public boolean acquire(long time, TimeUnit unit);
 
通過release()方法釋放鎖。InterProcessMutex 實例可以重用。Revoking ZooKeeper recipes wiki定義了可協(xié)商的撤銷機制。為了撤銷mutex, 調(diào)用下面的方法:
- /**
 - * 將鎖設(shè)為可撤銷的. 當(dāng)別的進程或線程想讓你釋放鎖時Listener會被調(diào)用。
 - * Parameters:
 - * listener - the listener
 - */
 - public void makeRevocable(RevocationListener<T> listener)
 
2)不可重入鎖Shared Lock
使用InterProcessSemaphoreMutex,調(diào)用方法類似,區(qū)別在于該鎖是不可重入的,在同一個線程中不可重入
3)可重入讀寫鎖Shared Reentrant Read Write Lock
類似JDK的ReentrantReadWriteLock. 一個讀寫鎖管理一對相關(guān)的鎖。一個負(fù)責(zé)讀操作,另外一個負(fù)責(zé)寫操作。讀操作在寫鎖沒被使用時可同時由多個進程使用,而寫鎖使用時不允許讀 (阻塞)。此鎖是可重入的。一個擁有寫鎖的線程可重入讀鎖,但是讀鎖卻不能進入寫鎖。這也意味著寫鎖可以降級成讀鎖, 比如請求寫鎖 —>讀鎖 —->釋放寫鎖。從讀鎖升級成寫鎖是不成的。主要由兩個類實現(xiàn):
- InterProcessReadWriteLock
 - InterProcessLock
 
4)信號量Shared Semaphore
一個計數(shù)的信號量類似JDK的Semaphore。JDK中Semaphore維護的一組許可(permits),而Cubator中稱之為租約(Lease)。注意,所有的實例必須使用相同的numberOfLeases值。調(diào)用acquire會返回一個租約對象??蛻舳吮仨氃趂inally中close這些租約對象,否則這些租約會丟失掉。但是, 但是,如果客戶端session由于某種原因比如crash丟掉, 那么這些客戶端持有的租約會自動close, 這樣其它客戶端可以繼續(xù)使用這些租約。租約還可以通過下面的方式返還:
- public void returnAll(Collection<Lease> leases)
 - public void returnLease(Lease lease)
 
注意一次你可以請求多個租約,如果Semaphore當(dāng)前的租約不夠,則請求線程會被阻塞。同時還提供了超時的重載方法:
- public Lease acquire()
 - public Collection<Lease> acquire(int qty)
 - public Lease acquire(long time, TimeUnit unit)
 - public Collection<Lease> acquire(int qty, long time, TimeUnit unit)
 
主要類有:
- InterProcessSemaphoreV2
 - Lease
 - SharedCountReader
 
5)多鎖對象Multi Shared Lock
Multi Shared Lock是一個鎖的容器。當(dāng)調(diào)用acquire, 所有的鎖都會被acquire,如果請求失敗,所有的鎖都會被release。同樣調(diào)用release時所有的鎖都被release(失敗被忽略)?;旧希褪墙M鎖的代表,在它上面的請求釋放操作都會傳遞給它包含的所有的鎖。主要涉及兩個類:
- InterProcessMultiLock
 - InterProcessLock
 
它的構(gòu)造函數(shù)需要包含的鎖的集合,或者一組ZooKeeper的path。
- public InterProcessMultiLock(List<InterProcessLock> locks)
 - public InterProcessMultiLock(CuratorFramework client, List<String> paths)
 
柵欄
barrier1)DistributedBarrier構(gòu)造函數(shù)中barrierPath參數(shù)用來確定一個柵欄,只要barrierPath參數(shù)相同(路徑相同)就是同一個柵欄。通常情況下柵欄的使用如下:
1.主導(dǎo)client設(shè)置一個柵欄
2.其他客戶端就會調(diào)用waitOnBarrier()等待柵欄移除,程序處理線程阻塞
3.主導(dǎo)client移除柵欄,其他客戶端的處理程序就會同時繼續(xù)運行。
DistributedBarrier類的主要方法如下:
setBarrier() - 設(shè)置柵欄
waitOnBarrier() - 等待柵欄移除
removeBarrier() - 移除柵欄
2)雙柵欄Double Barrier
雙柵欄允許客戶端在計算的開始和結(jié)束時同步。當(dāng)足夠的進程加入到雙柵欄時,進程開始計算,當(dāng)計算完成時,離開柵欄。雙柵欄類是DistributedDoubleBarrier DistributedDoubleBarrier類實現(xiàn)了雙柵欄的功能。它的構(gòu)造函數(shù)如下:
- // client - the client
 - // barrierPath - path to use
 - // memberQty - the number of members in the barrier
 - public DistributedDoubleBarrier(CuratorFramework client, String barrierPath, int memberQty)
 
memberQty是成員數(shù)量,當(dāng)enter方法被調(diào)用時,成員被阻塞,直到所有的成員都調(diào)用了enter。當(dāng)leave方法被調(diào)用時,它也阻塞調(diào)用線程,直到所有的成員都調(diào)用了leave。
注意:參數(shù)memberQty的值只是一個閾值,而不是一個限制值。當(dāng)?shù)却龞艡诘臄?shù)量大于或等于這個值柵欄就會打開!
與柵欄(DistributedBarrier)一樣,雙柵欄的barrierPath參數(shù)也是用來確定是否是同一個柵欄的,雙柵欄的使用情況如下:
1.從多個客戶端在同一個路徑上創(chuàng)建雙柵欄(DistributedDoubleBarrier),然后調(diào)用enter()方法,等待柵欄數(shù)量達(dá)到memberQty時就可以進入柵欄。
2.柵欄數(shù)量達(dá)到memberQty,多個客戶端同時停止阻塞繼續(xù)運行,直到執(zhí)行l(wèi)eave()方法,等待memberQty個數(shù)量的柵欄同時阻塞到leave()方法中。
3.memberQty個數(shù)量的柵欄同時阻塞到leave()方法中,多個客戶端的leave()方法停止阻塞,繼續(xù)運行。
DistributedDoubleBarrier類的主要方法如下:enter()、enter(long maxWait, TimeUnit unit) - 等待同時進入柵欄
leave()、leave(long maxWait, TimeUnit unit) - 等待同時離開柵欄
異常處理:DistributedDoubleBarrier會監(jiān)控連接狀態(tài),當(dāng)連接斷掉時enter()和leave方法會拋出異常。
計數(shù)器
Counters利用ZooKeeper可以實現(xiàn)一個集群共享的計數(shù)器。只要使用相同的path就可以得到最新的計數(shù)器值, 這是由ZooKeeper的一致性保證的。Curator有兩個計數(shù)器, 一個是用int來計數(shù),一個用long來計數(shù)。
1)SharedCount
這個類使用int類型來計數(shù)。主要涉及三個類。
- * SharedCount
 - * SharedCountReader
 - * SharedCountListener
 
SharedCount代表計數(shù)器, 可以為它增加一個SharedCountListener,當(dāng)計數(shù)器改變時此Listener可以監(jiān)聽到改變的事件,而SharedCountReader可以讀取到最新的值, 包括字面值和帶版本信息的值VersionedValue。
2)DistributedAtomicLong
除了計數(shù)的范圍比SharedCount大了之外, 它首先嘗試使用樂觀鎖的方式設(shè)置計數(shù)器, 如果不成功(比如期間計數(shù)器已經(jīng)被其它client更新了), 它使用InterProcessMutex方式來更新計數(shù)值。此計數(shù)器有一系列的操作:
- get(): 獲取當(dāng)前值
 - increment():加一
 - decrement(): 減一
 - add():增加特定的值
 - subtract(): 減去特定的值
 - trySet(): 嘗試設(shè)置計數(shù)值
 - forceSet(): 強制設(shè)置計數(shù)值
 
你必須檢查返回結(jié)果的succeeded(), 它代表此操作是否成功。如果操作成功, preValue()代表操作前的值, postValue()代表操作后的值。
End
Curator抽象和簡化了很多復(fù)雜的zookeeper操作,是zk使用者的福音。而要徹底的幸福,那就是不再使用它。
我不知道其他人把zk放在一個什么位置,但在我接觸paxos協(xié)議之后,就很難對它產(chǎn)生濃厚的興趣。一般在技術(shù)選型的時候,它會躺在我的備選列表最后,我甚至根本無法掌握源代碼里那些晦澀難懂的邏輯。
但工程建設(shè)從來不以我們的喜好來進行衡量。從來如此。
作者簡介:小姐姐味道 (xjjdog),一個不允許程序員走彎路的公眾號。聚焦基礎(chǔ)架構(gòu)和Linux。十年架構(gòu),日百億流量,與你探討高并發(fā)世界,給你不一樣的味道















 
 
 
 
 
 
 