Zookeeper?Curator使用介紹

更新時間：2022年09月02日 15:47:14 作者：悠然予夏

curator是Net?ix公司開源的?套Zookeeper客戶端框架，和ZKClient?樣，Curator解決了很多Zookeeper客戶端非常底層的細節(jié)開發(fā)?作，包括連接重連，反復注冊Watcher和NodeExistsException異常等，是最流行的Zookeeper客戶端之?

1、添加依賴

在pom.xml文件中添加如下內(nèi)容：

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>2.12.0</version>
</dependency>

2、創(chuàng)建會話

Curator的創(chuàng)建會話方式與原生的API和ZkClient的創(chuàng)建方式區(qū)別很?。Curator創(chuàng)建客戶端是通過CuratorFrameworkFactory工廠類來實現(xiàn)的。具體如下：

1. 使用CuratorFramework這個工廠類的兩個靜態(tài)方法來創(chuàng)建?個客戶端

public static CuratorFramework newClient(String connectString, RetryPolicy retryPolicy)
public static CuratorFramework newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)

其中參數(shù)RetryPolicy提供重試策略的接口，可以讓?戶實現(xiàn)?定義的重試策略，默認提供了以下實現(xiàn)，分別為ExponentialBacko?Retry（基于backo?的重連策略）、RetryNTimes（重連N次策略）、RetryForever（永遠重試策略）

2. 通過調(diào)用CuratorFramework中的start()方法來啟動會話

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3); 
CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181",retryPolicy); client.start();

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3); 
CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", 5000,1000,retryPolicy);
client.start();

其實進?步查看源代碼可以得知，其實這兩種方法內(nèi)部實現(xiàn)?樣，只是對外包裝成不同的方法。它們的底層都是通過第三個?法builder來實現(xiàn)的。

RetryPolicy retryPolicy	= new ExponentialBackoffRetry(1000,3);
private static CuratorFramework Client = CuratorFrameworkFactory.builder()
            .connectString("server1:2181,server2:2181,server3:2181")
            .sessionTimeoutMs(50000)
            .connectionTimeoutMs(30000)
            .retryPolicy(retryPolicy)
            .build();
client.start();

參數(shù)：

connectString：zk的server地址，多個server之間使?英?逗號分隔開

connectionTimeoutMs：連接超時時間，如上是 30s，默認是 15

ssessionTimeoutMs：會話超時時間，如上是 50s，默認是 60s

retryPolicy：失敗重試策略

ExponentialBacko?Retry：構(gòu)造器含有三個參數(shù) ExponentialBacko?Retry(int baseSleepTimeMs, int maxRetries, int maxSleepMs)

baseSleepTimeMs：初始的sleep時間，用于計算之后的每次重試的sleep時間

計算公式：當前 sleep 時間 =baseSleepTimeMs*Math.max(1, random.nextInt(1<<(retryCount+1)))

maxRetries：最大重試次數(shù)

maxSleepMs：最大sleep時間，如果上述的當前sleep計算出來比這個大，那么sleep用這個時間，默認的最大時間是Integer.MAX_VALUE毫秒。

其他，查看org.apache.curator.RetryPolicy接?的實現(xiàn)類

start()：完成會話的創(chuàng)建

package com.lagou.curator;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class CreateSession {
    // 創(chuàng)建會話
    public static void main(String[] args) {
        // 不使用fluent編程風格
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("8.142.8.105:2181", retryPolicy);
        curatorFramework.start();
        System.out.println("會話被建立了");
        // 使用fluent編程風格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("8.142.8.105:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("base")  // 獨立的命名空間 /base
                .build();
        client.start();
        System.out.println("會話2被創(chuàng)建了");
    }
}

需要注意的是session2會話含有隔離命名空間，即客戶端對Zookeeper上數(shù)據(jù)節(jié)點的任何操作都是相對/base目錄進行的，這有利于實現(xiàn)不同的Zookeeper的業(yè)務之間的隔離

3、創(chuàng)建節(jié)點

curator提供了?系列Fluent風格的接口，通過使用Fluent編程風格的接口，開發(fā)人員可以進行自由組合來完成各種類型節(jié)點的創(chuàng)建。

下面簡單介紹?下常用的幾個節(jié)點創(chuàng)建場景。

（1）創(chuàng)建?個初始內(nèi)容為空的節(jié)點

client.create().forPath(path);

Curator默認創(chuàng)建的是持久節(jié)點，內(nèi)容為空。

（2）創(chuàng)建?個包含內(nèi)容的節(jié)點

client.create().forPath(path,"我是內(nèi)容".getBytes());

Curator和ZkClient不同的是依舊采用Zookeeper原生API的?格，內(nèi)容使用byte[]作為方法參數(shù)。

（3）遞歸創(chuàng)建父節(jié)點,并選擇節(jié)點類型

client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);

creatingParentsIfNeeded這個接口非常有用，在使用ZooKeeper 的過程中，開發(fā)人員經(jīng)常會碰到NoNodeException 異常，其中?個可能的原因就是試圖對?個不存在的父節(jié)點創(chuàng)建子節(jié)點。因此，開發(fā)人員不得不在每次創(chuàng)建節(jié)點之前，都判斷?下該父節(jié)點是否存在——這個處理通常比較麻煩。在使用Curator 之后，通過調(diào)用creatingParentsIfNeeded接口，Curator就能夠自動地遞歸創(chuàng)建所有需要的?節(jié)點。

下面通過一個實際例子來演示如何在代碼中使用這些API。

package com.lagou.curator;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
public class CreateNote_curator {
    // 創(chuàng)建會話
    public static void main(String[] args) throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // 使用fluent編程風格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("8.142.8.105:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("base")  // 獨立的命名空間 /base
                .build();
        client.start();
        System.out.println("會話2被創(chuàng)建了");
        // 創(chuàng)建節(jié)點
        String path = "/lg-curator/c1";
        String s = client.create().creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT).forPath(path, "init".getBytes());
        System.out.println("節(jié)點遞歸創(chuàng)建成功，該節(jié)點路徑：" + s);
    }
}

4、刪除節(jié)點

刪除節(jié)點的方法也是基于Fluent方式來進行操作，不同類型的操作調(diào)用新增不同的方法調(diào)用即可。

（1）刪除一個子節(jié)點

client.delete().forPath(path);

（2）刪除節(jié)點并遞歸刪除其子節(jié)點

client.delete().deletingChildrenIfNeeded().forPath(path);

（3）指定版本進行刪除

client.delete().withVersion(1).forPath(path);

如果此版本已經(jīng)不存在，則刪除異常，異常信息如下。

org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode =BadVersion for

（4）強制保證刪除一個節(jié)點

client.delete().guaranteed().forPath(path);

只要客戶端會話有效，那么Curator會在后臺持續(xù)進行刪除操作，直到節(jié)點刪除成功。比如遇到?些網(wǎng)絡異常的情況，此guaranteed的強制刪除就會很有效果。

演示實例：

package com.lagou.curator;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class DeleteNote_curator {
    // 創(chuàng)建會話
    public static void main(String[] args) throws Exception {
        // 不使用fluent編程風格
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // 使用fluent編程風格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("8.142.8.105:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("base")  // 獨立的命名空間 /base
                .build();
        client.start();
        System.out.println("會話2被創(chuàng)建了");
        // 刪除節(jié)點
        String path = "/lg-curator";
        client.delete().deletingChildrenIfNeeded().withVersion(-1).forPath(path);
        System.out.println("刪除成功，刪除的節(jié)點：" + path);
    }
}

5、獲取數(shù)據(jù)

獲取節(jié)點數(shù)據(jù)內(nèi)容API相當簡單，同時Curator提供了傳入一個Stat變量的方式來存儲服務器端返回的最新的節(jié)點狀態(tài)信息

// 普通查詢
client.getData().forPath(path);
// 包含狀態(tài)查詢
Stat stat = new Stat(); 
client.getData().storingStatIn(stat).forPath(path);

演示：

package com.lagou.curator;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
public class GetNote_curator {
    // 創(chuàng)建會話
    public static void main(String[] args) throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // 使用fluent編程風格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("8.142.8.105:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("base")  // 獨立的命名空間 /base
                .build();
        client.start();
        System.out.println("會話2被創(chuàng)建了");
        // 創(chuàng)建節(jié)點
        String path = "/lg-curator/c1";
        String s = client.create().creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT).forPath(path, "init".getBytes());
        System.out.println("節(jié)點遞歸創(chuàng)建成功，該節(jié)點路徑：" + s);
        // 獲取節(jié)點的數(shù)據(jù)內(nèi)容以及狀態(tài)信息
        // 數(shù)據(jù)內(nèi)容
        byte[] bytes = client.getData().forPath(path);
        System.out.println("獲取到的節(jié)點數(shù)據(jù)內(nèi)容：" + new String(bytes));
        // 狀態(tài)信息
        Stat stat = new Stat();
        client.getData().storingStatIn(stat).forPath(path);
        System.out.println("獲取節(jié)點的狀態(tài)信息：" + stat);
    }
}

6、更新數(shù)據(jù)

更新數(shù)據(jù)，如果未傳入version參數(shù)，那么更新當前最新版本，如果傳入version則更新指定version，如果version已經(jīng)變更，則拋出異常。

// 普通更新
client.setData().forPath(path,"新內(nèi)容".getBytes());
// 指定版本更新
client.setData().withVersion(1).forPath(path);

版本不一致異常信息：

org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion for

案例演示：

package com.lagou.curator;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
public class UpdateNote_curator {
    // 創(chuàng)建會話
    public static void main(String[] args) throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // 使用fluent編程風格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("8.142.8.105:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("base")  // 獨立的命名空間 /base
                .build();
        client.start();
        System.out.println("會話2被創(chuàng)建了");
        String path = "/lg-curator/c1";
        // 獲取節(jié)點的數(shù)據(jù)內(nèi)容以及狀態(tài)信息
        // 數(shù)據(jù)內(nèi)容
        byte[] bytes = client.getData().forPath(path);
        System.out.println("獲取到的節(jié)點數(shù)據(jù)內(nèi)容：" + new String(bytes));
        // 狀態(tài)信息
        Stat stat = new Stat();
        client.getData().storingStatIn(stat).forPath(path);
        System.out.println("獲取節(jié)點的狀態(tài)信息：" + stat);
        // 更新節(jié)點內(nèi)容
        int version = client.setData().withVersion(stat.getVersion()).forPath(path, "修改內(nèi)容1".getBytes()).getVersion();
        System.out.println("當前的最新版本是"+version);
        byte[] byte2 = client.getData().forPath(path);
        System.out.println("修改后的節(jié)點數(shù)據(jù)內(nèi)容：" + new String(byte2));
    }
}

到此這篇關(guān)于Zookeeper Curator使用介紹的文章就介紹到這了,更多相關(guān)Zookeeper Curator內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: