本章代碼已分享至Gitee: https://gitee.com/lengcz/curator01

zookeeper 的安裝

關于zookeeper的安裝請見: dubbo(2):zookeeper和dubbo-admin的安裝

Curator 介紹

Curator 是Apache Zookeeper 的java 客戶端庫
常見的Zookeeper Java API

• 原生Java API
• ZkClient
• Curator

Curator 項目的目標是簡化Zookeeper 客戶端的使用。
Curator 最初是Netfix 研發的，后來捐獻給了Apache基金會，目前屬于 Apache頂級項目。
Curator 官網： https://curator.apache.org/

Curator API 常用操作

• 建立連接
• 添加節點
• 刪除節點
• 修改節點
• 查詢節點
• Watch事件監聽
• 分布式鎖實現

本章必要的相關依賴和配置

<dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.10</version><scope>test</scope></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>4.0.0</version></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>4.0.0</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.21</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.21</version></dependency><!-- https://mvnrepository.com/artifact/com.alibaba.fastjson2/fastjson2 --><dependency><groupId>com.alibaba.fastjson2</groupId><artifactId>fastjson2</artifactId><version>2.0.57</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.1</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin></plugins></build>

建立連接

第一種方式

        //1 第一種方式/** String connectString, 連接字符串 zk server 地址和端口 ip1:port1,ip2:port2....*  int sessionTimeoutMs, 會話超時時間  單位ms* int connectionTimeoutMs,  連接超時時間,單位ms*  RetryPolicy retryPolicy  重拾策略*/RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);String connectString = "localhost:2181";CuratorFramework client =CuratorFrameworkFactory.newClient(connectString, 60_000, 15_000, retryPolicy);client.start();

第二種方式

//2 第二種方式,鏈式編程CuratorFramework client2 = CuratorFrameworkFactory.builder().connectString(connectString).sessionTimeoutMs(60_000).connectionTimeoutMs(15_000).retryPolicy(retryPolicy).namespace("demo01").build();client2.start();

namespace表示根節點，表示這個客戶端的操作都在這個節點之下，后續使用中不需要從根開始聲明。

• 節點不需要手動創建，連接是會自動創建

關閉連接

 client.close();

添加節點

創建節點

@Testpublic void testCreate() throws Exception {//如果創建節點，沒有指定數據，則默認將當前客戶端的ip作為數據存儲String path = client.create().forPath("/app1");logger.info(path);}

創建節點并設置值和類型

 @Testpublic void testCreate() throws Exception {//如果創建節點，沒有指定數據，則默認將當前客戶端的ip作為數據存儲
//        String path = client.create().forPath("/app1");
//        logger.info(path);// withMode 指定節點類型，是臨時的，還是臨時的，順序的String path2 = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3","hello world".getBytes()); //創建節點并設置值logger.info(path2);}

withMode ：模式。EPHEMERAL 臨時的，當會話關閉時，節點會被刪除。

創建多級節點

  // creatingParentContainersIfNeeded 如果節點不存在則創建String path3 = client.create().creatingParentContainersIfNeeded().forPath("/app4/p1"); //創建多級節點logger.info(path3);

• creatingParentContainersIfNeeded 如果節點不存在則創建

查詢節點

查詢數據

  byte[] bytes = client.getData().forPath("/app1");logger.info(new String(bytes));

查詢所有子節點

//獲取子節點List<String> paths = client.getChildren().forPath("/app4");paths.forEach(c ->logger.info(c));List<String> paths2 = client.getChildren().forPath("/");//查詢根節點

查詢節點信息

@Testpublic void testGetNodeInfo() throws Exception {//  查詢節點狀態信息 ls -sStat status = new Stat();client.getData().storingStatIn(status).forPath("/app1");logger.info(JSONObject.toJSONString(status));}

修改節點

修改節點數據

  @Testpublic void testSet() throws Exception {// 修改數據client.setData().forPath("/app1","hello".getBytes());}

修改節點(樂觀鎖修改，根據版本號)

 @Testpublic void testSetForVersion() throws Exception {Stat status = new Stat();//定義空的Stat接受返回結果client.getData().storingStatIn(status).forPath("/app1");int version = status.getVersion();//查詢版本logger.info("版本號:"+version);// 修改數據,如果版本不一致，則不修改數據，樂觀鎖模式client.setData().withVersion(version).forPath("/app1","hello".getBytes());}

圖片表示版本號

創建或者修改數據

如果節點不存在，則創建節點，并設置數據，如果節點存在則修改數據

@Testpublic void testCreatOrSetDataForVersion() throws Exception {Stat status = new Stat();//定義空的Stat接受返回結果client.create().orSetData.storingStatIn(status).forPath("/app1");int version = status.getVersion();//查詢版本logger.info("版本號:"+version);}

刪除節點

刪除節點

  /*** 1 刪除節點 delete* 2 刪除節點帶有子節點的節點 deleteall* 3 必須成功的刪除* 4 回調* @throws Exception*/@Testpublic void testDelete() throws Exception {// 刪除節點dclient.delete().forPath("/app1");}

刪除帶有子節點的節點

  @Testpublic void testDelete2() throws Exception {// 刪除節點帶有子節點的節點client.delete().deletingChildrenIfNeeded().forPath("/app4");}

必須刪除成功

 @Testpublic void testDelete3() throws Exception {// 必須刪除成功client.delete().guaranteed().forPath("/app4");}

必須成功就是重試，防止網絡抖動

刪除后回調

 @Testpublic void testCallback() throws Exception{// 回調client.delete().guaranteed().inBackground(new BackgroundCallback() {@Overridepublic void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {logger.info("刪除后的消息回調");}}).forPath("/app4");}