偷偷摘套内射激情视频,久久精品99国产国产精,中文字幕无线乱码人妻,中文在线中文a,性爽19p

ZK客戶端Curator使用詳解

作者：小姐姐養(yǎng)的狗
開(kāi)源
zookeeper不是為高可用性設(shè)計(jì)的，但它使用ZAB協(xié)議達(dá)到了極高的一致性，所以是個(gè)CP系統(tǒng)。所以它經(jīng)常被選作注冊(cè)中心、配置中心、分布式鎖等場(chǎng)景。

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

[[323374]]

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

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

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

1.maven依賴

最新版本的curator 4.3.0支持zookeeper 3.4.x和3.5，但是需要注意curator傳遞進(jìn)來(lái)的依賴，需要和實(shí)際服務(wù)器端使用的版本相符，以我們目前使用的zookeeper 3.4.6為例。

  1. <dependency> 
  2.     <groupId>org.apache.curator</groupId> 
  3.     <artifactId>curator-framework</artifactId> 
  4.     <version>4.3.0</version> 
  5.     <exclusions> 
  6.         <exclusion> 
  7.             <groupId>org.apache.zookeeper</groupId> 
  8.             <artifactId>zookeeper</artifactId> 
  9.         </exclusion> 
  10.     </exclusions> 
  11. </dependency> 
  12. <dependency> 
  13.     <groupId>org.apache.curator</groupId> 
  14.     <artifactId>curator-recipes</artifactId> 
  15.     <version>4.3.0</version> 
  16.     <exclusions> 
  17.         <exclusion> 
  18.             <groupId>org.apache.zookeeper</groupId> 
  19.             <artifactId>zookeeper</artifactId> 
  20.         </exclusion> 
  21.     </exclusions> 
  22. </dependency> 
  23. <dependency> 
  24.     <groupId>org.apache.zookeeper</groupId> 
  25.     <artifactId>zookeeper</artifactId> 
  26.     <version>3.4.6</version> 
  27. </dependency> 

2.curator-framework

下面是一些常見(jiàn)的zk相關(guān)的操作。

  1. public static CuratorFramework getClient() { 
  2.     return CuratorFrameworkFactory.builder() 
  3.             .connectString("127.0.0.1:2181") 
  4.             .retryPolicy(new ExponentialBackoffRetry(1000, 3)) 
  5.             .connectionTimeoutMs(15 * 1000) //連接超時(shí)時(shí)間，默認(rèn)15秒 
  6.             .sessionTimeoutMs(60 * 1000) //會(huì)話超時(shí)時(shí)間，默認(rèn)60秒 
  7.             .namespace("arch") //設(shè)置命名空間 
  8.             .build(); 
  9. } 
  10.   
  11. public static void create(final CuratorFramework client, final String path, final byte[] payload) throws Exception { 
  12.     client.create().creatingParentsIfNeeded().forPath(path, payload); 
  13. } 
  14.   
  15. public static void createEphemeral(final CuratorFramework client, final String path, final byte[] payload) throws Exception { 
  16.     client.create().withMode(CreateMode.EPHEMERAL).forPath(path, payload); 
  17. } 
  18.   
  19. public static String createEphemeralSequential(final CuratorFramework client, final String path, final byte[] payload) throws Exception { 
  20.     return client.create().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, payload); 
  21. } 
  22.   
  23. public static void setData(final CuratorFramework client, final String path, final byte[] payload) throws Exception { 
  24.     client.setData().forPath(path, payload); 
  25. } 
  26.   
  27. public static void delete(final CuratorFramework client, final String path) throws Exception { 
  28.     client.delete().deletingChildrenIfNeeded().forPath(path); 
  29. } 
  30.   
  31. public static void guaranteedDelete(final CuratorFramework client, final String path) throws Exception { 
  32.     client.delete().guaranteed().forPath(path); 
  33. } 
  34.   
  35. public static String getData(final CuratorFramework client, final String path) throws Exception { 
  36.     return new String(client.getData().forPath(path)); 
  37. } 
  38.   
  39. public static List<String> getChildren(final CuratorFramework client, final String path) throws Exception { 
  40.     return client.getChildren().forPath(path); 
  41. } 

3.curator-recipescurator-recipes

提供了一些zk的典型使用場(chǎng)景的參考。下面主要介紹一下開(kāi)發(fā)中常用的組件。

事件監(jiān)聽(tīng)

zookeeper原生支持通過(guò)注冊(cè)watcher來(lái)進(jìn)行事件監(jiān)聽(tīng)，但是其使用不是特別方便，需要開(kāi)發(fā)人員自己反復(fù)注冊(cè)watcher，比較繁瑣。

Curator引入Cache來(lái)實(shí)現(xiàn)對(duì)zookeeper服務(wù)端事務(wù)的監(jiān)聽(tīng)。Cache是Curator中對(duì)事件監(jiān)聽(tīng)的包裝，其對(duì)事件的監(jiān)聽(tīng)其實(shí)可以近似看作是一個(gè)本地緩存視圖和遠(yuǎn)程Zookeeper視圖的對(duì)比過(guò)程。同時(shí)，Curator能夠自動(dòng)為開(kāi)發(fā)人員處理反復(fù)注冊(cè)監(jiān)聽(tīng)，從而大大簡(jiǎn)化原生api開(kāi)發(fā)的繁瑣過(guò)程。

1)Node Cache

  1. public static void nodeCache() throws Exception { 
  2.     final String path = "/nodeCache"; 
  3.     final CuratorFramework client = getClient(); 
  4.     client.start(); 
  5.   
  6.     delete(client, path); 
  7.     create(client, path, "cache".getBytes()); 
  8.   
  9.     final NodeCache nodeCache = new NodeCache(client, path); 
  10.     nodeCache.start(true); 
  11.     nodeCache.getListenable() 
  12.             .addListener(() -> System.out.println("node data change, new data is " + new String(nodeCache.getCurrentData().getData()))); 
  13.   
  14.     setData(client, path, "cache1".getBytes()); 
  15.     setData(client, path, "cache2".getBytes()); 
  16.   
  17.     Thread.sleep(1000); 
  18.   
  19.     client.close(); 
  20. } 

NodeCache可以監(jiān)聽(tīng)指定的節(jié)點(diǎn)，注冊(cè)監(jiān)聽(tīng)器后，節(jié)點(diǎn)的變化會(huì)通知相應(yīng)的監(jiān)聽(tīng)器

2)Path Cache

Path Cache 用來(lái)監(jiān)聽(tīng)ZNode的子節(jié)點(diǎn)事件，包括added、updateed、removed，Path Cache會(huì)同步子節(jié)點(diǎn)的狀態(tài)，產(chǎn)生的事件會(huì)傳遞給注冊(cè)的PathChildrenCacheListener。

  1. public static void pathChildrenCache() throws Exception { 
  2.         final String path = "/pathChildrenCache"; 
  3.         final CuratorFramework client = getClient(); 
  4.         client.start(); 
  5.   
  6.         final PathChildrenCache cache = new PathChildrenCache(client, path, true); 
  7.         cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT); 
  8.     cache.getListenable().addListener((client1, event) -> { 
  9.             switch (event.getType()) { 
  10.                 case CHILD_ADDED: 
  11.                     System.out.println("CHILD_ADDED:" + event.getData().getPath()); 
  12.                     break; 
  13.                 case CHILD_REMOVED: 
  14.                     System.out.println("CHILD_REMOVED:" + event.getData().getPath()); 
  15.                     break; 
  16.                 case CHILD_UPDATED: 
  17.                     System.out.println("CHILD_UPDATED:" + event.getData().getPath()); 
  18.                     break; 
  19.                 case CONNECTION_LOST: 
  20.                     System.out.println("CONNECTION_LOST:" + event.getData().getPath()); 
  21.                     break; 
  22.                 case CONNECTION_RECONNECTED: 
  23.                     System.out.println("CONNECTION_RECONNECTED:" + event.getData().getPath()); 
  24.                     break; 
  25.                 case CONNECTION_SUSPENDED: 
  26.                     System.out.println("CONNECTION_SUSPENDED:" + event.getData().getPath()); 
  27.                     break; 
  28.                 case INITIALIZED: 
  29.                     System.out.println("INITIALIZED:" + event.getData().getPath()); 
  30.                     break; 
  31.                 default: 
  32.                     break; 
  33.             } 
  34.         }); 
  35.   
  36. //        client.create().withMode(CreateMode.PERSISTENT).forPath(path); 
  37.         Thread.sleep(1000); 
  38.   
  39.         client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1"); 
  40.         Thread.sleep(1000); 
  41.   
  42.         client.delete().forPath(path + "/c1"); 
  43.         Thread.sleep(1000); 
  44.   
  45.         client.delete().forPath(path); //監(jiān)聽(tīng)節(jié)點(diǎn)本身的變化不會(huì)通知 
  46.         Thread.sleep(1000); 
  47.   
  48.         client.close(); 
  49.     } 

3)Tree Cache

Path Cache和Node Cache的“合體”，監(jiān)視路徑下的創(chuàng)建、更新、刪除事件，并緩存路徑下所有孩子結(jié)點(diǎn)的數(shù)據(jù)。

  1. public static void treeCache() throws Exception { 
  2.     final String path = "/treeChildrenCache"; 
  3.     final CuratorFramework client = getClient(); 
  4.     client.start(); 
  5.   
  6.     final TreeCache cache = new TreeCache(client, path); 
  7.     cache.start(); 
  8.   
  9.     cache.getListenable().addListener((client1, event) -> { 
  10.         switch (event.getType()){ 
  11.             case NODE_ADDED: 
  12.                 System.out.println("NODE_ADDED:" + event.getData().getPath()); 
  13.                 break; 
  14.             case NODE_REMOVED: 
  15.                 System.out.println("NODE_REMOVED:" + event.getData().getPath()); 
  16.                 break; 
  17.             case NODE_UPDATED: 
  18.                 System.out.println("NODE_UPDATED:" + event.getData().getPath()); 
  19.                 break; 
  20.             case CONNECTION_LOST: 
  21.                 System.out.println("CONNECTION_LOST:" + event.getData().getPath()); 
  22.                 break; 
  23.             case CONNECTION_RECONNECTED: 
  24.                 System.out.println("CONNECTION_RECONNECTED:" + event.getData().getPath()); 
  25.                 break; 
  26.             case CONNECTION_SUSPENDED: 
  27.                 System.out.println("CONNECTION_SUSPENDED:" + event.getData().getPath()); 
  28.                 break; 
  29.             case INITIALIZED: 
  30.                 System.out.println("INITIALIZED:" + event.getData().getPath()); 
  31.                 break; 
  32.             default: 
  33.                 break; 
  34.         } 
  35.     }); 
  36.   
  37.     client.create().withMode(CreateMode.PERSISTENT).forPath(path); 
  38.     Thread.sleep(1000); 
  39.   
  40.     client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1"); 
  41.     Thread.sleep(1000); 
  42.   
  43.     setData(client, path, "test".getBytes()); 
  44.     Thread.sleep(1000); 
  45.   
  46.     client.delete().forPath(path + "/c1"); 
  47.     Thread.sleep(1000); 
  48.   
  49.     client.delete().forPath(path); 
  50.     Thread.sleep(1000); 
  51.   
  52.     client.close(); 
  53. } 

選舉

curator提供了兩種方式，分別是Leader Latch和Leader Election。

1)Leader Latch

隨機(jī)從候選著中選出一臺(tái)作為leader，選中之后除非調(diào)用close()釋放leadship，否則其他的后選擇無(wú)法成為leader

  1. public class LeaderLatchTest { 
  2.   
  3.     private static final String PATH = "/demo/leader"; 
  4.   
  5.     public static void main(String[] args) { 
  6.         List<LeaderLatch> latchList = new ArrayList<>(); 
  7.         List<CuratorFramework> clients = new ArrayList<>(); 
  8.         try { 
  9.             for (int i = 0; i < 10; i++) { 
  10.                 CuratorFramework client = getClient(); 
  11.                 client.start(); 
  12.                 clients.add(client); 
  13.   
  14.                 final LeaderLatch leaderLatch = new LeaderLatch(client, PATH, "client#" + i); 
  15.                 leaderLatch.addListener(new LeaderLatchListener() { 
  16.                     @Override 
  17.                     public void isLeader() { 
  18.                         System.out.println(leaderLatch.getId() + ":I am leader. I am doing jobs!"); 
  19.                     } 
  20.   
  21.                     @Override 
  22.                     public void notLeader() { 
  23.                         System.out.println(leaderLatch.getId() + ":I am not leader. I will do nothing!"); 
  24.                     } 
  25.                 }); 
  26.                 latchList.add(leaderLatch); 
  27.                 leaderLatch.start(); 
  28.             } 
  29.             Thread.sleep(1000 * 60); 
  30.         } catch (Exception e) { 
  31.             e.printStackTrace(); 
  32.         } finally { 
  33.             for (CuratorFramework client : clients) { 
  34.                 CloseableUtils.closeQuietly(client); 
  35.             } 
  36.   
  37.             for (LeaderLatch leaderLatch : latchList) { 
  38.                 CloseableUtils.closeQuietly(leaderLatch); 
  39.             } 
  40.         } 
  41.     } 
  42.   
  43.     public static CuratorFramework getClient() { 
  44.         return CuratorFrameworkFactory.builder() 
  45.                 .connectString("127.0.0.1:2181") 
  46.                 .retryPolicy(new ExponentialBackoffRetry(1000, 3)) 
  47.                 .connectionTimeoutMs(15 * 1000) //連接超時(shí)時(shí)間，默認(rèn)15秒 
  48.                 .sessionTimeoutMs(60 * 1000) //會(huì)話超時(shí)時(shí)間，默認(rèn)60秒 
  49.                 .namespace("arch") //設(shè)置命名空間 
  50.                 .build(); 
  51.     } 
  52.   
  53. } 

2)Leader Election

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

  1. public class LeaderSelectorTest { 
  2.     private static final String PATH = "/demo/leader"; 
  3.   
  4.     public static void main(String[] args) { 
  5.         List<LeaderSelector> selectors = new ArrayList<>(); 
  6.         List<CuratorFramework> clients = new ArrayList<>(); 
  7.         try { 
  8.             for (int i = 0; i < 10; i++) { 
  9.                 CuratorFramework client = getClient(); 
  10.                 client.start(); 
  11.                 clients.add(client); 
  12.   
  13.                 final String name = "client#" + i; 
  14.                 LeaderSelector leaderSelector = new LeaderSelector(client, PATH, new LeaderSelectorListenerAdapter() { 
  15.                     @Override 
  16.                     public void takeLeadership(CuratorFramework client) throws Exception { 
  17.                         System.out.println(name + ":I am leader."); 
  18.                         Thread.sleep(2000); 
  19.                     } 
  20.                 }); 
  21.   
  22.                 leaderSelector.autoRequeue(); 
  23.                 leaderSelector.start(); 
  24.                 selectors.add(leaderSelector); 
  25.             } 
  26.             Thread.sleep(Integer.MAX_VALUE); 
  27.         } catch (Exception e) { 
  28.             e.printStackTrace(); 
  29.         } finally { 
  30.             for (CuratorFramework client : clients) { 
  31.                 CloseableUtils.closeQuietly(client); 
  32.             } 
  33.   
  34.             for (LeaderSelector selector : selectors) { 
  35.                 CloseableUtils.closeQuietly(selector); 
  36.             } 
  37.   
  38.         } 
  39.     } 
  40.   
  41.     public static CuratorFramework getClient() { 
  42.         return CuratorFrameworkFactory.builder() 
  43.                 .connectString("127.0.0.1:2181") 
  44.                 .retryPolicy(new ExponentialBackoffRetry(1000, 3)) 
  45.                 .connectionTimeoutMs(15 * 1000) //連接超時(shí)時(shí)間，默認(rèn)15秒 
  46.                 .sessionTimeoutMs(60 * 1000) //會(huì)話超時(shí)時(shí)間，默認(rèn)60秒 
  47.                 .namespace("arch") //設(shè)置命名空間 
  48.                 .build(); 
  49.     } 
  50.   
  51. } 

分布式鎖

1)可重入鎖Shared Reentrant Lock

Shared意味著鎖是全局可見(jiàn)的， 客戶端都可以請(qǐng)求鎖。Reentrant和JDK的ReentrantLock類似， 意味著同一個(gè)客戶端在擁有鎖的同時(shí)，可以多次獲取，不會(huì)被阻塞。它是由類InterProcessMutex來(lái)實(shí)現(xiàn)。它的構(gòu)造函數(shù)為：

  1. public InterProcessMutex(CuratorFramework client, String path) 

通過(guò)acquire獲得鎖，并提供超時(shí)機(jī)制：

  1. /** 
  2. * Acquire the mutex - blocking until it's available. Note: the same thread can call acquire 
  3. * re-entrantly. Each call to acquire must be balanced by a call to release() 
  4. */ 
  5. public void acquire(); 
  6.   
  7. /** 
  8. * Acquire the mutex - blocks until it's available or the given time expires. Note: the same thread can 
  9. * call acquire re-entrantly. Each call to acquire that returns true must be balanced by a call to release() 
  10. * Parameters: 
  11. * time - time to wait 
  12. * unit - time unit 
  13. * Returns: 
  14. * true if the mutex was acquired, false if not 
  15. */ 
  16. public boolean acquire(long time, TimeUnit unit); 

通過(guò)release()方法釋放鎖。InterProcessMutex 實(shí)例可以重用。Revoking ZooKeeper recipes wiki定義了可協(xié)商的撤銷機(jī)制。為了撤銷mutex, 調(diào)用下面的方法：

  1. /** 
  2. * 將鎖設(shè)為可撤銷的. 當(dāng)別的進(jìn)程或線程想讓你釋放鎖時(shí)Listener會(huì)被調(diào)用。 
  3. * Parameters: 
  4. * listener - the listener 
  5. */ 
  6. public void makeRevocable(RevocationListener<T> listener) 

2)不可重入鎖Shared Lock

使用InterProcessSemaphoreMutex，調(diào)用方法類似，區(qū)別在于該鎖是不可重入的，在同一個(gè)線程中不可重入

3)可重入讀寫(xiě)鎖Shared Reentrant Read Write Lock

類似JDK的ReentrantReadWriteLock. 一個(gè)讀寫(xiě)鎖管理一對(duì)相關(guān)的鎖。一個(gè)負(fù)責(zé)讀操作，另外一個(gè)負(fù)責(zé)寫(xiě)操作。讀操作在寫(xiě)鎖沒(méi)被使用時(shí)可同時(shí)由多個(gè)進(jìn)程使用，而寫(xiě)鎖使用時(shí)不允許讀 (阻塞)。此鎖是可重入的。一個(gè)擁有寫(xiě)鎖的線程可重入讀鎖，但是讀鎖卻不能進(jìn)入寫(xiě)鎖。這也意味著寫(xiě)鎖可以降級(jí)成讀鎖， 比如請(qǐng)求寫(xiě)鎖 —>讀鎖 —->釋放寫(xiě)鎖。從讀鎖升級(jí)成寫(xiě)鎖是不成的。主要由兩個(gè)類實(shí)現(xiàn)：

  1. InterProcessReadWriteLock 
  2. InterProcessLock 

4)信號(hào)量Shared Semaphore

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

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

注意一次你可以請(qǐng)求多個(gè)租約，如果Semaphore當(dāng)前的租約不夠，則請(qǐng)求線程會(huì)被阻塞。同時(shí)還提供了超時(shí)的重載方法:

  1. public Lease acquire() 
  2. public Collection<Lease> acquire(int qty) 
  3. public Lease acquire(long time, TimeUnit unit) 
  4. public Collection<Lease> acquire(int qty, long time, TimeUnit unit) 

主要類有:

  1. InterProcessSemaphoreV2 
  2. Lease 
  3. SharedCountReader 

5)多鎖對(duì)象Multi Shared Lock

Multi Shared Lock是一個(gè)鎖的容器。當(dāng)調(diào)用acquire， 所有的鎖都會(huì)被acquire，如果請(qǐng)求失敗，所有的鎖都會(huì)被release。同樣調(diào)用release時(shí)所有的鎖都被release(失敗被忽略)?；旧?，它就是組鎖的代表，在它上面的請(qǐng)求釋放操作都會(huì)傳遞給它包含的所有的鎖。主要涉及兩個(gè)類：

  1. InterProcessMultiLock 
  2. InterProcessLock 

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

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

柵欄

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

1.主導(dǎo)client設(shè)置一個(gè)柵欄

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

3.主導(dǎo)client移除柵欄，其他客戶端的處理程序就會(huì)同時(shí)繼續(xù)運(yùn)行。

DistributedBarrier類的主要方法如下：

setBarrier() - 設(shè)置柵欄

waitOnBarrier() - 等待柵欄移除

removeBarrier() - 移除柵欄

2)雙柵欄Double Barrier

雙柵欄允許客戶端在計(jì)算的開(kāi)始和結(jié)束時(shí)同步。當(dāng)足夠的進(jìn)程加入到雙柵欄時(shí)，進(jìn)程開(kāi)始計(jì)算，當(dāng)計(jì)算完成時(shí)，離開(kāi)柵欄。雙柵欄類是DistributedDoubleBarrier DistributedDoubleBarrier類實(shí)現(xiàn)了雙柵欄的功能。它的構(gòu)造函數(shù)如下：

  1. // client - the client 
  2. // barrierPath - path to use 
  3. // memberQty - the number of members in the barrier 
  4. public DistributedDoubleBarrier(CuratorFramework client, String barrierPath, int memberQty) 

memberQty是成員數(shù)量，當(dāng)enter方法被調(diào)用時(shí)，成員被阻塞，直到所有的成員都調(diào)用了enter。當(dāng)leave方法被調(diào)用時(shí)，它也阻塞調(diào)用線程，直到所有的成員都調(diào)用了leave。

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

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

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

2.柵欄數(shù)量達(dá)到memberQty，多個(gè)客戶端同時(shí)停止阻塞繼續(xù)運(yùn)行，直到執(zhí)行l(wèi)eave()方法，等待memberQty個(gè)數(shù)量的柵欄同時(shí)阻塞到leave()方法中。

3.memberQty個(gè)數(shù)量的柵欄同時(shí)阻塞到leave()方法中，多個(gè)客戶端的leave()方法停止阻塞，繼續(xù)運(yùn)行。

DistributedDoubleBarrier類的主要方法如下：enter()、enter(long maxWait, TimeUnit unit) - 等待同時(shí)進(jìn)入柵欄

leave()、leave(long maxWait, TimeUnit unit) - 等待同時(shí)離開(kāi)柵欄

異常處理：DistributedDoubleBarrier會(huì)監(jiān)控連接狀態(tài)，當(dāng)連接斷掉時(shí)enter()和leave方法會(huì)拋出異常。

計(jì)數(shù)器

Counters利用ZooKeeper可以實(shí)現(xiàn)一個(gè)集群共享的計(jì)數(shù)器。只要使用相同的path就可以得到最新的計(jì)數(shù)器值， 這是由ZooKeeper的一致性保證的。Curator有兩個(gè)計(jì)數(shù)器， 一個(gè)是用int來(lái)計(jì)數(shù)，一個(gè)用long來(lái)計(jì)數(shù)。

1)SharedCount

這個(gè)類使用int類型來(lái)計(jì)數(shù)。主要涉及三個(gè)類。

  1. * SharedCount 
  2. * SharedCountReader 
  3. * SharedCountListener 

SharedCount代表計(jì)數(shù)器， 可以為它增加一個(gè)SharedCountListener，當(dāng)計(jì)數(shù)器改變時(shí)此Listener可以監(jiān)聽(tīng)到改變的事件，而SharedCountReader可以讀取到最新的值， 包括字面值和帶版本信息的值VersionedValue。

2)DistributedAtomicLong

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

  • get(): 獲取當(dāng)前值
  • increment()：加一
  • decrement(): 減一
  • add()：增加特定的值
  • subtract(): 減去特定的值
  • trySet(): 嘗試設(shè)置計(jì)數(shù)值
  • forceSet(): 強(qiáng)制設(shè)置計(jì)數(shù)值

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

End

Curator抽象和簡(jiǎn)化了很多復(fù)雜的zookeeper操作，是zk使用者的福音。而要徹底的幸福，那就是不再使用它。

我不知道其他人把zk放在一個(gè)什么位置，但在我接觸paxos協(xié)議之后，就很難對(duì)它產(chǎn)生濃厚的興趣。一般在技術(shù)選型的時(shí)候，它會(huì)躺在我的備選列表最后，我甚至根本無(wú)法掌握源代碼里那些晦澀難懂的邏輯。

但工程建設(shè)從來(lái)不以我們的喜好來(lái)進(jìn)行衡量。從來(lái)如此。

作者簡(jiǎn)介：小姐姐味道 (xjjdog)，一個(gè)不允許程序員走彎路的公眾號(hào)。聚焦基礎(chǔ)架構(gòu)和Linux。十年架構(gòu)，日百億流量，與你探討高并發(fā)世界，給你不一樣的味道

責(zé)任編輯：武曉燕 來(lái)源： 小姐姐味道
zookeeperCP系統(tǒng)
相關(guān)推薦

2010-05-12 15:46:51

Subversion客

2010-06-01 14:11:11

TortoiseSVN

2011-08-17 10:10:59

2009-03-04 10:27:50

客戶端組件桌面虛擬化Xendesktop

2010-04-08 15:35:13

Oracle 簡(jiǎn)易客戶

2012-10-11 17:02:02

IBMdw

2011-03-21 14:53:36

Nagios監(jiān)控Linux

2011-04-06 14:24:20

Nagios監(jiān)控Linux

2009-07-24 17:31:56

ASP.NET AJA

2010-05-31 15:55:42

2013-06-08 09:59:15

VMwarevSphere Web

2010-03-18 16:49:43

Java Socket

2025-08-28 08:26:47

2012-01-13 10:29:37

ibmdw

2010-06-01 13:54:42

TortoiseSVN

2011-04-06 14:24:27

Nagios監(jiān)控Linux

2010-05-26 09:26:43

Cassandra

2009-08-21 16:14:52

服務(wù)端與客戶端通信

2021-04-30 08:19:32

SpringCloud客戶端負(fù)載Ribbo

2010-03-18 15:44:22

VSTS 2010VS 2010
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號(hào)