zookeeper客戶端Curator怎么使用

這篇文章主要講解了“zookeeper客戶端Curator怎么使用”，文中的講解內(nèi)容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“zookeeper客戶端Curator怎么使用”吧！

zookeeper不是為高可用性設計的，但它使用ZAB協(xié)議達到了極高的一致性，所以是個CP系統(tǒng)。所以它經(jīng)常被選作注冊中心、配置中心、分布式鎖等場景。

它的性能是非常有限的，而且API并不是那么好用。xjjdog傾向于使用基于Raft協(xié)議的Etcd或者Consul，它們更加輕量級一些。

Curator是netflix公司開源的一套zookeeper客戶端，目前是Apache的頂級項目。與Zookeeper提供的原生客戶端相比，Curator的抽象層次更高，簡化了Zookeeper客戶端的開發(fā)量。Curator解決了很多zookeeper客戶端非常底層的細節(jié)開發(fā)工作，包括連接重連、反復注冊wathcer和NodeExistsException  異常等。

Curator由一系列的模塊構成，對于一般開發(fā)者而言，常用的是curator-framework和curator-recipes，下面對此依次介紹。

1.maven依賴

最新版本的curator 4.3.0支持zookeeper  3.4.x和3.5，但是需要注意curator傳遞進來的依賴，需要和實際服務器端使用的版本相符，以我們目前使用的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相關的操作。

public static CuratorFramework getClient() {     return CuratorFrameworkFactory.builder()             .connectString("127.0.0.1:2181")             .retryPolicy(new ExponentialBackoffRetry(1000, 3))             .connectionTimeoutMs(15 * 1000) //連接超時時間，默認15秒             .sessionTimeoutMs(60 * 1000) //會話超時時間，默認60秒             .namespace("arch") //設置命名空間             .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ā)人員自己反復注冊watcher，比較繁瑣。

Curator引入Cache來實現(xiàn)對zookeeper服務端事務的監(jiān)聽。Cache是Curator中對事件監(jiān)聽的包裝，其對事件的監(jiān)聽其實可以近似看作是一個本地緩存視圖和遠程Zookeeper視圖的對比過程。同時，Curator能夠自動為開發(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é)點的變化會通知相應的監(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)建、更新、刪除事件，并緩存路徑下所有孩子結點的數(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) //連接超時時間，默認15秒                 .sessionTimeoutMs(60 * 1000) //會話超時時間，默認60秒                 .namespace("arch") //設置命名空間                 .build();     }   }

2)Leader Election

通過LeaderSelectorListener可以對領導權進行控制，  在適當?shù)臅r候釋放領導權，這樣每個節(jié)點都有可能獲得領導權。而LeaderLatch則一直持有l(wèi)eadership，  除非調(diào)用close方法，否則它不會釋放領導權。

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) //連接超時時間，默認15秒                 .sessionTimeoutMs(60 * 1000) //會話超時時間，默認60秒                 .namespace("arch") //設置命名空間                 .build();     }   }

分布式鎖

1)可重入鎖Shared Reentrant Lock

Shared意味著鎖是全局可見的， 客戶端都可以請求鎖。Reentrant和JDK的ReentrantLock類似，  意味著同一個客戶端在擁有鎖的同時，可以多次獲取，不會被阻塞。它是由類InterProcessMutex來實現(xiàn)。它的構造函數(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)用下面的方法：

/** * 將鎖設為可撤銷的. 當別的進程或線程想讓你釋放鎖時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.  一個讀寫鎖管理一對相關的鎖。一個負責讀操作，另外一個負責寫操作。讀操作在寫鎖沒被使用時可同時由多個進程使用，而寫鎖使用時不允許讀  (阻塞)。此鎖是可重入的。一個擁有寫鎖的線程可重入讀鎖，但是讀鎖卻不能進入寫鎖。這也意味著寫鎖可以降級成讀鎖， 比如請求寫鎖 —>讀鎖  —->釋放寫鎖。從讀鎖升級成寫鎖是不成的。主要由兩個類實現(xiàn)：

InterProcessReadWriteLock InterProcessLock

4)信號量Shared Semaphore

一個計數(shù)的信號量類似JDK的Semaphore。JDK中Semaphore維護的一組許可(permits)，而Cubator中稱之為租約(Lease)。注意，所有的實例必須使用相同的numberOfLeases值。調(diào)用acquire會返回一個租約對象。客戶端必須在finally中close這些租約對象，否則這些租約會丟失掉。但是，  但是，如果客戶端session由于某種原因比如crash丟掉， 那么這些客戶端持有的租約會自動close，  這樣其它客戶端可以繼續(xù)使用這些租約。租約還可以通過下面的方式返還：

public void returnAll(Collection<Lease> leases) public void returnLease(Lease lease)

注意一次你可以請求多個租約，如果Semaphore當前的租約不夠，則請求線程會被阻塞。同時還提供了超時的重載方法:

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是一個鎖的容器。當調(diào)用acquire，  所有的鎖都會被acquire，如果請求失敗，所有的鎖都會被release。同樣調(diào)用release時所有的鎖都被release(失敗被忽略)。基本上，它就是組鎖的代表，在它上面的請求釋放操作都會傳遞給它包含的所有的鎖。主要涉及兩個類：

InterProcessMultiLock InterProcessLock

它的構造函數(shù)需要包含的鎖的集合，或者一組ZooKeeper的path。

public InterProcessMultiLock(List<InterProcessLock> locks) public InterProcessMultiLock(CuratorFramework client, List<String> paths)

柵欄

barrier1)DistributedBarrier構造函數(shù)中barrierPath參數(shù)用來確定一個柵欄，只要barrierPath參數(shù)相同(路徑相同)就是同一個柵欄。通常情況下柵欄的使用如下：

1.主導client設置一個柵欄

2.其他客戶端就會調(diào)用waitOnBarrier()等待柵欄移除，程序處理線程阻塞

3.主導client移除柵欄，其他客戶端的處理程序就會同時繼續(xù)運行。

DistributedBarrier類的主要方法如下：

setBarrier() - 設置柵欄

waitOnBarrier() - 等待柵欄移除

removeBarrier() - 移除柵欄

2)雙柵欄Double Barrier

雙柵欄允許客戶端在計算的開始和結束時同步。當足夠的進程加入到雙柵欄時，進程開始計算，當計算完成時，離開柵欄。雙柵欄類是DistributedDoubleBarrier  DistributedDoubleBarrier類實現(xiàn)了雙柵欄的功能。它的構造函數(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ù)量，當enter方法被調(diào)用時，成員被阻塞，直到所有的成員都調(diào)用了enter。當leave方法被調(diào)用時，它也阻塞調(diào)用線程，直到所有的成員都調(diào)用了leave。

注意：參數(shù)memberQty的值只是一個閾值，而不是一個限制值。當?shù)却龞艡诘臄?shù)量大于或等于這個值柵欄就會打開!

與柵欄(DistributedBarrier)一樣,雙柵欄的barrierPath參數(shù)也是用來確定是否是同一個柵欄的，雙柵欄的使用情況如下：

1.從多個客戶端在同一個路徑上創(chuàng)建雙柵欄(DistributedDoubleBarrier),然后調(diào)用enter()方法，等待柵欄數(shù)量達到memberQty時就可以進入柵欄。

2.柵欄數(shù)量達到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)，當連接斷掉時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，當計數(shù)器改變時此Listener可以監(jiān)聽到改變的事件，而SharedCountReader可以讀取到最新的值，  包括字面值和帶版本信息的值VersionedValue。

2)DistributedAtomicLong

除了計數(shù)的范圍比SharedCount大了之外， 它首先嘗試使用樂觀鎖的方式設置計數(shù)器， 如果不成功(比如期間計數(shù)器已經(jīng)被其它client更新了)，  它使用InterProcessMutex方式來更新計數(shù)值。此計數(shù)器有一系列的操作：

• get(): 獲取當前值

• increment()：加一

• decrement(): 減一

• add()：增加特定的值

• subtract(): 減去特定的值

• trySet(): 嘗試設置計數(shù)值

• forceSet(): 強制設置計數(shù)值

你必須檢查返回結果的succeeded()， 它代表此操作是否成功。如果操作成功， preValue()代表操作前的值，  postValue()代表操作后的值。

感謝各位的閱讀，以上就是“zookeeper客戶端Curator怎么使用”的內(nèi)容了，經(jīng)過本文的學習后，相信大家對zookeeper客戶端Curator怎么使用這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！