#zookeeper

Part1原文

https://www.baeldung.com/apache-curator

11. Introduction

Apache Curator?is a Java client for?Apache Zookeeper, the popular coordination service for distributed applications.

Apache Curator 是 Apache Zookeeper 的 Java 客戶端，后者是分布式應(yīng)用程序的流行協(xié)調(diào)服務(wù)。

In this tutorial, we'll introduce some of the most relevant features provided by Curator:

• Connection Management – managing connections and retry policies

• Async – enhancing existing client by adding async capabilities and the use of Java 8 lambdas

• Configuration Management – having a centralized configuration for the system

• Strongly-Typed Models – working with typed models

• Recipes – implementing leader election, distributed locks or counters

在本文中，我們講介紹一些Curator提供的最實(shí)用的功能實(shí)現(xiàn)：

• 連接管理：管理連接和重試策略

• 異步：通過(guò)添加異步功能和使用 Java 8 lambda 來(lái)增強(qiáng)現(xiàn)有客戶端

• 配置管理：系統(tǒng)集中配置。

• 強(qiáng)類型模型：作用于類型模型上

• Recipes：實(shí)現(xiàn)領(lǐng)導(dǎo)選舉，分布式鎖或者計(jì)數(shù)器

22. Prerequisites

To start with, it’s recommended to take a quick look at the?Apache Zookeeper?and its features.

首先，建議快速瀏覽一下 Apache Zookeeper 及其功能。

For this tutorial, we assume that there's already a standalone Zookeeper instance running on?127.0.0.1:2181;?here are?instructions on how to install and run it, if you're just getting started.

對(duì)于本教程，我們假設(shè)已經(jīng)有一個(gè)獨(dú)立的 Zookeeper 實(shí)例在 127.0.0.1:2181 上運(yùn)行；這里 會(huì)介紹如何安裝并且運(yùn)行

[[【Zookeeper】基于3臺(tái)linux虛擬機(jī)搭建zookeeper集群]]

First, we’ll need to add the?curator-x-async?dependency to our?pom.xml:

首先，我們需要添加 [curator-x-async](https://search.maven.org/classic/#search%7Cgav%7C1%7Cg%3A%22org.apache.curator%22%20AND% 20a%3A%22curator-x-async%22) 對(duì)我們的 pom.xml 的依賴：

<dependency>
????<groupId>org.apache.curator</groupId>
????<artifactId>curator-x-async</artifactId>
????<version>4.0.1</version>
????<exclusions>
????????<exclusion>
????????????<groupId>org.apache.zookeeper</groupId>
????????????<artifactId>zookeeper</artifactId>
????????</exclusion>
????</exclusions>
</dependency>


The latest version of Apache Curator 4.X.X has a hard dependency with Zookeeper 3.5.X?which is still in beta right now.

最新版本的 Apache Curator 4.X.X 與 Zookeeper 3.5.X 具有硬依賴性。

And so, in this article, we’re going to use the currently latest stable?Zookeeper 3.4.11?instead.

在本文中會(huì)使用文章編寫時(shí)候的最后一個(gè)穩(wěn)定版本 Zookeeper 3.4.11 替代

So we need to exclude the Zookeeper dependency and add?the dependency for our Zookeeper version?to our?pom.xml:

因此，我們需要排除 Zookeeper 依賴項(xiàng)并添加[我們的 Zookeeper 版本的依賴項(xiàng)](https://search.maven.org/classic/#search%7Cgav%7C1%7Cg%3A%22org.apache.zookeeper%22% 20AND%20a%3A%22zookeeper%22) 到我們的 _pom.xml_：

<dependency>
????<groupId>org.apache.zookeeper</groupId>
????<artifactId>zookeeper</artifactId>
????<version>3.4.11</version>
</dependency>


For more information about compatibility, please refer to?this link.

更多信息可以查看：?this link

33. Connection Management

The basic use case of Apache Curator is connecting to a running Apache Zookeeper instance.

最基礎(chǔ)的使用案例，是使用Curator連接并且運(yùn)行一個(gè)Apach Zk 實(shí)例。

The tool provides a factory to build connections to Zookeeper using retry policies:

工具提供了一個(gè)工廠構(gòu)建ZK連接，并且需要手動(dòng)指定重試器。

int?sleepMsBetweenRetries?=?100;
int?maxRetries?=?3;
RetryPolicy?retryPolicy?=?new?RetryNTimes(
??maxRetries,?sleepMsBetweenRetries);

CuratorFramework?client?=?CuratorFrameworkFactory
??.newClient("127.0.0.1:2181",?retryPolicy);
client.start();
?
assertThat(client.checkExists().forPath("/")).isNotNull();


In this quick example, we'll retry 3 times and will wait 100 ms between retries in case of connectivity issues.

在這個(gè)快速使用案例中，我們指定了重試三次，并且每次重試的間隔等待100ms。

Once connected to Zookeeper using the?CuratorFramework?client, we can now browse paths, get/set data and essentially interact with the server.

使用 CuratorFramework 客戶端連接到 Zookeeper 后，我們現(xiàn)在可以瀏覽路徑、獲取/設(shè)置數(shù)據(jù)并與服務(wù)器進(jìn)行交互。

44. Async

The Curator Async module wraps the above?CuratorFramework?client to provide non-blocking capabilities?using?the CompletionStage Java 8 API.

Curator Async 模塊使用CompletionStage Java 8 API 對(duì)上述 CuratorFramework 客戶端進(jìn)行封裝，以提供非阻塞功能。

Let's see how the previous example looks like using the Async wrapper:

讓我們看看使用 Async 封裝器的前一個(gè)示例是怎樣的：

int?sleepMsBetweenRetries?=?100;
int?maxRetries?=?3;
RetryPolicy?retryPolicy?
??=?new?RetryNTimes(maxRetries,?sleepMsBetweenRetries);

CuratorFramework?client?=?CuratorFrameworkFactory
??.newClient("127.0.0.1:2181",?retryPolicy);

client.start();
AsyncCuratorFramework?async?=?AsyncCuratorFramework.wrap(client);

AtomicBoolean?exists?=?new?AtomicBoolean(false);

async.checkExists()
??.forPath("/")
??.thenAcceptAsync(s?->?exists.set(s?!=?null));

await().until(()?->?assertThat(exists.get()).isTrue());


Now, the?checkExists()?operation works in asynchronous mode, not blocking the main thread. We can also chain actions one after each other using the?thenAcceptAsync()?method instead, which uses the?CompletionStage API.

現(xiàn)在，_checkExists()_ 操作以異步模式運(yùn)行，不會(huì)阻塞主線程。我們還可以使用_thenAcceptAsync()_方法（該方法使用 CompletionStage API）將操作一個(gè)接一個(gè)地串聯(lián)起來(lái)。

55. Configuration Management

In a distributed environment, one of the most common challenges is to manage shared configuration among many applications.?We can use Zookeeper as a data store where to keep our configuration.

在分布式環(huán)境中，最常見(jiàn)的挑戰(zhàn)之一就是管理多個(gè)應(yīng)用程序之間的共享配置。?**我們可以使用 Zookeeper 作為保存配置的數(shù)據(jù)存儲(chǔ)。

Let's see an example using Apache Curator to get and set data:

讓我們來(lái)看一個(gè)使用 Apache Curator 獲取和設(shè)置數(shù)據(jù)的示例：

CuratorFramework?client?=?newClient();
client.start();
AsyncCuratorFramework?async?=?AsyncCuratorFramework.wrap(client);
String?key?=?getKey();
String?expected?=?"my_value";

client.create().forPath(key);

async.setData()
??.forPath(key,?expected.getBytes());

AtomicBoolean?isEquals?=?new?AtomicBoolean();
async.getData()
??.forPath(key)
??.thenAccept(data?->?isEquals.set(new?String(data).equals(expected)));

await().until(()?->?assertThat(isEquals.get()).isTrue());


In this example, we create the node path, set the data in Zookeeper, and then we recover it checking that the value is the same. The?key?field could be a node path like?/config/dev/my_key.

在這個(gè)示例中，我們創(chuàng)建節(jié)點(diǎn)路徑，在 Zookeeper 中設(shè)置數(shù)據(jù)，然后檢查值是否相同，再恢復(fù)數(shù)據(jù)。_key_ 字段可以是一個(gè)節(jié)點(diǎn)路徑，如 _/config/dev/my_key_。

5.1. Watchers

Another interesting feature in Zookeeper is the ability to watch keys or nodes.?It allows us to listen to changes in the configuration and update our applications without needing to redeploy.

Zookeeper 的另一個(gè)有趣功能是監(jiān)視鍵或節(jié)點(diǎn)。?它允許我們監(jiān)聽(tīng)配置中的變化并更新應(yīng)用程序，而無(wú)需重新部署。

Let's see how the above example looks like when using watchers:

讓我們看看上面的示例在使用Watch時(shí)是什么樣子的：

CuratorFramework?client?=?newClient()
client.start();
AsyncCuratorFramework?async?=?AsyncCuratorFramework.wrap(client);
String?key?=?getKey();
String?expected?=?"my_value";

async.create().forPath(key);

List<String>?changes?=?new?ArrayList<>();

async.watched()
??.getData()
??.forPath(key)
??.event()
??.thenAccept(watchedEvent?->?{
????try?{
????????changes.add(new?String(client.getData()
??????????.forPath(watchedEvent.getPath())));
????}?catch?(Exception?e)?{
????????//?fail?...
????}});

//?Set?data?value?for?our?key
async.setData()
??.forPath(key,?expected.getBytes());

await()
??.until(()?->?assertThat(changes.size()).isEqualTo(1));


We configure the watcher, set the data, and then confirm the watched event was triggered. We can watch one node or a set of nodes at once.

我們配置監(jiān)視器、設(shè)置數(shù)據(jù)，然后確認(rèn)被監(jiān)視事件已觸發(fā)。我們可以同時(shí)觀察一個(gè)節(jié)點(diǎn)或一組節(jié)點(diǎn)。

66. Strongly Typed Models

Zookeeper primarily works with byte arrays, so we need to serialize and deserialize our data. This allows us some flexibility to work with any serializable instance, but it can be hard to maintain.

Zookeeper 主要使用字節(jié)數(shù)組，因此我們需要對(duì)數(shù)據(jù)進(jìn)行序列化和反序列化。這讓我們可以靈活地使用任何可序列化的實(shí)例，但也很難維護(hù)。

To help here, Curator adds the concept of?typed models?which?delegates the serialization/deserialization and allows us to work with our types directly. Let's see how that works.

為了在這方面提供幫助，Curator 添加了類型化模型的概念，它了序列化/反序列化，并允許我們直接我們的類型。讓我們看看它是如何工作的。

First, we need a serializer framework. Curator recommends using the Jackson implementation, so let's add?the Jackson dependency?to our?pom.xml:

首先，我們需要一個(gè)序列化框架。Curator 推薦使用 Jackson 實(shí)現(xiàn)，因此讓我們?cè)?pom.xml 中添加Jackson 依賴項(xiàng)：

<dependency>
????<groupId>com.fasterxml.jackson.core</groupId>
????<artifactId>jackson-databind</artifactId>
????<version>2.13.0</version>
</dependency>


Now, let's try to persist our custom class?HostConfig:

現(xiàn)在，讓我們嘗試持久化自定義類 _HostConfig_：

public?class?HostConfig?{
????private?String?hostname;
????private?int?port;

????//?getters?and?setters
}


We need to provide the model specification mapping from the?HostConfig?class to a path, and use the modeled framework wrapper provided by Apache Curator:

我們需要提供從 HostConfig 類到路徑的模型規(guī)范映射，并使用 Apache Curator 提供的建模框架包裝器：

ModelSpec<HostConfig>?mySpec?=?ModelSpec.builder(
??ZPath.parseWithIds("/config/dev"),?
??JacksonModelSerializer.build(HostConfig.class))
??.build();

CuratorFramework?client?=?newClient();
client.start();

AsyncCuratorFramework?async?
??=?AsyncCuratorFramework.wrap(client);
ModeledFramework<HostConfig>?modeledClient?
??=?ModeledFramework.wrap(async,?mySpec);

modeledClient.set(new?HostConfig("host-name",?8080));

modeledClient.read()
??.whenComplete((value,?e)?->?{
?????if?(e?!=?null)?{
??????????fail("Cannot?read?host?config",?e);
?????}?else?{
??????????assertThat(value).isNotNull();
??????????assertThat(value.getHostname()).isEqualTo("host-name");
??????????assertThat(value.getPort()).isEqualTo(8080);
?????}
???});


The?whenComplete()?method when reading the path?/config/dev?will return the?HostConfig?instance in Zookeeper.

讀取 /config/dev 路徑時(shí)，_whenComplete()_ 方法將返回 Zookeeper 中的 HostConfig 實(shí)例。

77. Recipes

Zookeeper provides?this guideline?to implement?high-level solutions or recipes such as leader election, distributed locks or shared counters.

Zookeeper 提供了本指南來(lái)實(shí)現(xiàn)高層次的解決方案或秘訣，如領(lǐng)導(dǎo)者選舉、分布式鎖或共享計(jì)數(shù)器。

Apache Curator provides an implementation for most of these recipes. To see the full list, visit?the Curator Recipes documentation.

Apache Curator 提供了其中大部分方案的實(shí)現(xiàn)方法。要查看完整列表，請(qǐng)?jiān)L問(wèn)Curator Recipes 文檔。

All of these recipes are available in a separate module:

所有這些API都可以在單獨(dú)的模塊中找到：

<dependency>
????<groupId>org.apache.curator</groupId>
????<artifactId>curator-recipes</artifactId>
????<version>4.0.1</version>
</dependency>


Let's jump right in and start understanding these with some simple examples.

讓我們通過(guò)一些簡(jiǎn)單的例子，直接開(kāi)始了解這些知識(shí)。

7.1. Leader Election

In a distributed environment, we may need one master or leader node to coordinate a complex job.

在分布式環(huán)境中，我們可能需要一個(gè)主節(jié)點(diǎn)或領(lǐng)導(dǎo)節(jié)點(diǎn)來(lái)協(xié)調(diào)復(fù)雜的工作。

This is how the usage of?the Leader Election recipe?in Curator looks like:

這就是 領(lǐng)導(dǎo)人選舉配方 在 Curator 中的用法：

CuratorFramework?client?=?newClient();
client.start();
LeaderSelector?leaderSelector?=?new?LeaderSelector(client,?
??"/mutex/select/leader/for/job/A",?
??new?LeaderSelectorListener()?{
??????@Override
??????public?void?stateChanged(
????????CuratorFramework?client,?
????????ConnectionState?newState)?{
??????}

??????@Override
??????public?void?takeLeadership(
????????CuratorFramework?client)?throws?Exception?{
??????}
??});

//?join?the?members?group
leaderSelector.start();

//?wait?until?the?job?A?is?done?among?all?members
leaderSelector.close();


When we start the leader selector, our node joins a members group within the path?/mutex/select/leader/for/job/A. Once our node becomes the leader, the?takeLeadership?method will be invoked, and we as leaders can resume the job.

當(dāng)我們啟動(dòng)領(lǐng)導(dǎo)者選擇器時(shí)，我們的節(jié)點(diǎn)會(huì)加入 /mutex/select/leader/for/job/A 路徑下的成員組。一旦我們的節(jié)點(diǎn)成為領(lǐng)導(dǎo)者，_takeLeadership_ 方法就會(huì)被調(diào)用，作為領(lǐng)導(dǎo)者的我們就可以繼續(xù)執(zhí)行任務(wù)。

7.2. Shared Locks

The Shared Lock recipe?is about having a fully distributed lock:

共享鎖配方 是關(guān)于完全分布式鎖的：

CuratorFramework?client?=?newClient();
client.start();
InterProcessSemaphoreMutex?sharedLock?=?new?InterProcessSemaphoreMutex(
??client,?"/mutex/process/A");

sharedLock.acquire();

//?do?process?A

sharedLock.release();


When we acquire the lock, Zookeeper ensures that there's no other application acquiring the same lock at the same time.

當(dāng)我們獲取鎖時(shí)，Zookeeper 會(huì)確保沒(méi)有其他應(yīng)用程序同時(shí)獲取相同的鎖。

7.3. Counters

The Counters recipe?coordinates a shared?Integer?among all the clients:

計(jì)數(shù)器配方 協(xié)調(diào)所有客戶端共享的_整數(shù)：

CuratorFramework?client?=?newClient();
client.start();

SharedCount?counter?=?new?SharedCount(client,?"/counters/A",?0);
counter.start();

counter.setCount(counter.getCount()?+?1);

assertThat(counter.getCount()).isEqualTo(1);


In this example, Zookeeper stores the?Integer?value in the path?/counters/A?and initializes the value to?0?if the path has not been created yet.

在此示例中，Zookeeper 將 Integer 值存儲(chǔ)在 /counters/A 路徑中，如果路徑尚未創(chuàng)建，則將該值初始化為 _0_。

88. Conclusion

In this article, we’ve seen how to use Apache Curator to connect to Apache Zookeeper and take benefit of its main features.

在這篇文章中，我們介紹了如何使用Apach Curator連接Apach Zookeeper，并利用其主要功能。

We’ve also introduced a few of the main recipes in Curator.

我們還介紹了 Curator 中的一些主要不見(jiàn)。

As usual, sources can be found?over on GitHub.

與往常一樣，您可以在 GitHub 上找到源代碼。