Zookeeper?Curator使用介紹

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

curator是Net?ix公司開(kāi)源的?套Zookeeper客戶端框架，和ZKClient?樣，Curator解決了很多Zookeeper客戶端非常底層的細(xì)節(jié)開(kāi)發(fā)?作，包括連接重連，反復(fù)注冊(cè)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)建會(huì)話

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

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

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

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

2. 通過(guò)調(diào)用CuratorFramework中的start()方法來(lái)啟動(dòng)會(huì)話

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();

其實(shí)進(jìn)?步查看源代碼可以得知，其實(shí)這兩種方法內(nèi)部實(shí)現(xiàn)?樣，只是對(duì)外包裝成不同的方法。它們的底層都是通過(guò)第三個(gè)?法builder來(lái)實(shí)現(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地址，多個(gè)server之間使?英?逗號(hào)分隔開(kāi)

connectionTimeoutMs：連接超時(shí) 時(shí) 間，如上是 30s，默認(rèn) 是 15

ssessionTimeoutMs：會(huì) 話超時(shí) 時(shí) 間，如上是 50s，默認(rèn) 是 60s

retryPolicy：失敗重試策略

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

baseSleepTimeMs：初始的sleep時(shí)間，用于計(jì)算之后的每次重試的sleep時(shí)間

計(jì) 算公式：當(dāng) 前 sleep 時(shí) 間 =baseSleepTimeMs*Math.max(1, random.nextInt(1<<(retryCount+1)))

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

maxSleepMs：最大sleep時(shí)間，如果上述的當(dāng)前sleep計(jì)算出來(lái)比這個(gè)大，那么sleep用這個(gè)時(shí)間，默認(rèn)的最大時(shí)間是Integer.MAX_VALUE毫秒。

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

start()：完成會(huì)話的創(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)建會(huì)話
    public static void main(String[] args) {
        // 不使用fluent編程風(fēng)格
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("8.142.8.105:2181", retryPolicy);
        curatorFramework.start();
        System.out.println("會(huì)話被建立了");
        // 使用fluent編程風(fēng)格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("8.142.8.105:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("base")  // 獨(dú)立的命名空間 /base
                .build();
        client.start();
        System.out.println("會(huì)話2被創(chuàng)建了");
    }
}

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

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

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

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

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

client.create().forPath(path);

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

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

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

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

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

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

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

下面通過(guò)一個(gè)實(shí)際例子來(lái)演示如何在代碼中使用這些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)建會(huì)話
    public static void main(String[] args) throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // 使用fluent編程風(fēng)格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("8.142.8.105:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("base")  // 獨(dú)立的命名空間 /base
                .build();
        client.start();
        System.out.println("會(huì)話2被創(chuàng)建了");
        // 創(chuàng)建節(jié)點(diǎn)
        String path = "/lg-curator/c1";
        String s = client.create().creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT).forPath(path, "init".getBytes());
        System.out.println("節(jié)點(diǎn)遞歸創(chuàng)建成功，該節(jié)點(diǎn)路徑：" + s);
    }
}

4、刪除節(jié)點(diǎn)

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

（1）刪除一個(gè)子節(jié)點(diǎn)

client.delete().forPath(path);

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

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

（3）指定版本進(jìn)行刪除

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

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

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

（4）強(qiáng)制保證刪除一個(gè)節(jié)點(diǎn)

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

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

演示實(shí)例：

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)建會(huì)話
    public static void main(String[] args) throws Exception {
        // 不使用fluent編程風(fēng)格
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // 使用fluent編程風(fēng)格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("8.142.8.105:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("base")  // 獨(dú)立的命名空間 /base
                .build();
        client.start();
        System.out.println("會(huì)話2被創(chuàng)建了");
        // 刪除節(jié)點(diǎn)
        String path = "/lg-curator";
        client.delete().deletingChildrenIfNeeded().withVersion(-1).forPath(path);
        System.out.println("刪除成功，刪除的節(jié)點(diǎn)：" + path);
    }
}

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

獲取節(jié)點(diǎn)數(shù)據(jù)內(nèi)容API相當(dāng)簡(jiǎn)單，同時(shí)Curator提供了傳入一個(gè)Stat變量的方式來(lái)存儲(chǔ)服務(wù)器端返回的最新的節(jié)點(diǎn)狀態(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)建會(huì)話
    public static void main(String[] args) throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // 使用fluent編程風(fēng)格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("8.142.8.105:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("base")  // 獨(dú)立的命名空間 /base
                .build();
        client.start();
        System.out.println("會(huì)話2被創(chuàng)建了");
        // 創(chuàng)建節(jié)點(diǎn)
        String path = "/lg-curator/c1";
        String s = client.create().creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT).forPath(path, "init".getBytes());
        System.out.println("節(jié)點(diǎn)遞歸創(chuàng)建成功，該節(jié)點(diǎn)路徑：" + s);
        // 獲取節(jié)點(diǎn)的數(shù)據(jù)內(nèi)容以及狀態(tài)信息
        // 數(shù)據(jù)內(nèi)容
        byte[] bytes = client.getData().forPath(path);
        System.out.println("獲取到的節(jié)點(diǎn)數(shù)據(jù)內(nèi)容：" + new String(bytes));
        // 狀態(tài)信息
        Stat stat = new Stat();
        client.getData().storingStatIn(stat).forPath(path);
        System.out.println("獲取節(jié)點(diǎn)的狀態(tài)信息：" + stat);
    }
}

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

更新數(shù)據(jù)，如果未傳入version參數(shù)，那么更新當(dāng)前最新版本，如果傳入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)建會(huì)話
    public static void main(String[] args) throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // 使用fluent編程風(fēng)格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("8.142.8.105:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("base")  // 獨(dú)立的命名空間 /base
                .build();
        client.start();
        System.out.println("會(huì)話2被創(chuàng)建了");
        String path = "/lg-curator/c1";
        // 獲取節(jié)點(diǎn)的數(shù)據(jù)內(nèi)容以及狀態(tài)信息
        // 數(shù)據(jù)內(nèi)容
        byte[] bytes = client.getData().forPath(path);
        System.out.println("獲取到的節(jié)點(diǎn)數(shù)據(jù)內(nèi)容：" + new String(bytes));
        // 狀態(tài)信息
        Stat stat = new Stat();
        client.getData().storingStatIn(stat).forPath(path);
        System.out.println("獲取節(jié)點(diǎn)的狀態(tài)信息：" + stat);
        // 更新節(jié)點(diǎn)內(nèi)容
        int version = client.setData().withVersion(stat.getVersion()).forPath(path, "修改內(nèi)容1".getBytes()).getVersion();
        System.out.println("當(dāng)前的最新版本是"+version);
        byte[] byte2 = client.getData().forPath(path);
        System.out.println("修改后的節(jié)點(diǎn)數(shù)據(jù)內(nèi)容：" + new String(byte2));
    }
}

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

您可能感興趣的文章: