詳解Zookeeper基礎(chǔ)知識
更新時間：2021年09月01日 09:54:41 作者：張鐵牛
本文主要講解了Zookeeper的基礎(chǔ)知識，ZooKeeper提供了一個通用協(xié)調(diào)模式實現(xiàn)方法的開源共享庫，使程序員免于寫這類通用的協(xié)議。關(guān)于Zookeeper更多相關(guān)知識，感興趣的小伙伴參考一下這篇文章
1. 簡介
zookeeper是一個開源的分布式協(xié)調(diào)服務(wù), 提供分布式數(shù)據(jù)一致性解決方案,分布式應(yīng)用程序可以實現(xiàn)數(shù)據(jù)統(tǒng)一配置管理、統(tǒng)一命名服務(wù)、分布式鎖、集群管理等功能.
ZooKeeper主要服務(wù)于分布式系統(tǒng)，使用分布式系統(tǒng)就無法避免對節(jié)點管理的問題(需要實時感知節(jié)點的狀態(tài)、對節(jié)點進行統(tǒng)一管理等等)，而由于這些問題處理起來可能相對麻煩和提高了系統(tǒng)的復(fù)雜性，ZooKeeper作為一個能夠通用解決這些問題的中間件就應(yīng)運而生了。
2. 數(shù)據(jù)模型
2.1 模型結(jié)構(gòu)
ZooKeeper 提供的命名空間很像標準文件系統(tǒng)的命名空間。名稱是由斜杠 (/) 分隔的一系列路徑元素。ZooKeeper 命名空間中的每個節(jié)點都由路徑標識。
與典型的為存儲而設(shè)計的文件系統(tǒng)不同，ZooKeeper 數(shù)據(jù)保存在內(nèi)存中，這意味著 ZooKeeper 可以實現(xiàn)高吞吐量和低延遲。
2.2 模型的特點
- 每個子目錄如
/app1都被稱作一個znode（節(jié)點）。這個znode是被它所在的路徑唯一標識。 - znode可以有子節(jié)點目錄，并且每個znode可以存儲數(shù)據(jù)。
- znode是有版本的，每個znode中存儲的數(shù)據(jù)可以有多個版本，也就是一個訪問路徑中可以存儲多份數(shù)據(jù)。每次 znode 的數(shù)據(jù)更改時，版本號都會增加。例如，每當客戶端檢索數(shù)據(jù)時，它也會收到數(shù)據(jù)的版本。
- 每個znode都有一個訪問控制列表 (ACL)，它限制了誰可以做什么。
- znode可以被監(jiān)控，包括這個目錄中存儲的數(shù)據(jù)的修改，子節(jié)點目錄變化等，一旦變化可以通知設(shè)置監(jiān)控的客戶端。
⚠️注意：當使用zkCli.sh 創(chuàng)建會話時，節(jié)點的監(jiān)聽事件只能被觸發(fā)一次。
2.3 節(jié)點分類
2.3.1 Persistent
持久節(jié)點：節(jié)點被創(chuàng)建后，就一直存在，除非客戶端主動刪除這個節(jié)點。
2.3.2 Persistent Sequential
持久順序節(jié)點：有序的持久節(jié)點。在zk中，每個父節(jié)點會為它的第一級子節(jié)點維護一份時序，會記錄每個子節(jié)點創(chuàng)建的先后順序。例如：
$ create -s /app1 # 創(chuàng)建 /app1 節(jié)點 Created /app10000000000 # 創(chuàng)建成功后的節(jié)點名稱為 /app10000000000
2.3.3 Ephemeral
臨時節(jié)點：和持久節(jié)點不同的是，臨時節(jié)點的生命周期和客戶端會話綁定且臨時節(jié)點下面不能創(chuàng)建子節(jié)點。如果客戶端會話失效，那么這個節(jié)點就會自動被清除掉。
注意：客戶端失效臨時節(jié)點會被清除，但如果是斷開鏈接，臨時節(jié)點并不會立馬被清除。
“立馬”：在會話超時持續(xù)時間內(nèi)未從客戶端收到任何心跳信號之后，zk服務(wù)器將刪除該會話的臨時節(jié)點。但如果正常關(guān)閉會話，臨時節(jié)點會立馬被清除。
# 1. 創(chuàng)建會話 $ ./zkCli.sh # 2. 創(chuàng)建臨時節(jié)點 /app3 $ create -e /app3 Created /app3 # 3. ctrl + c 關(guān)閉會話 # 4. 緊接著再次創(chuàng)建會話 $ ./zkCli.sh # 5. 查看節(jié)點內(nèi)容 發(fā)現(xiàn)臨時節(jié)點依舊存在 $ ls / [app10000000000, app3, zookeeper] # 6. 等待幾秒，再次查看 發(fā)現(xiàn)臨時節(jié)點消失 $ ls / [app10000000000, zookeeper] # 7. 再次創(chuàng)建臨時節(jié)點 /app3 $ create -e /app3 Created /app3 # 8. 正常關(guān)閉會話 $ quit # 9. 再次創(chuàng)建會話 臨時節(jié)點消失 $ ./zkCli.sh $ ls / [app10000000000, zookeeper]
2.3.4 Ephemeral Sequential
臨時順序節(jié)點：有序的臨時節(jié)點。創(chuàng)建es節(jié)點時，zk會維護一份時序，會記錄每個節(jié)點的順序。例如：
$ create -s -e /app2 # 創(chuàng)建臨時有序節(jié)點 /app2 Created /app20000000001 # 創(chuàng)建成功后的節(jié)點名稱為 app20000000001
3. 安裝
3.1 官方
官方地址：https://zookeeper.apache.org/releases.html
3.2 docker
$ docker run --name zookeeper --restart always -d zookeeper
3.3 docker-compose
⚠️注意：執(zhí)行腳本前，需先將配置文件掛載到宿主機上。
version: '3.1'
services:
zk:
image: zookeeper
restart: always
container_name: zookeeper
ports:
- 2181:2181
volumes:
- ./data:/data
- ./logs:/datalog
- ./conf/zoo.cfg:/conf/zoo.cfg
3.4 配置信息
# 數(shù)據(jù)存儲目錄 dataDir=/data # 日志存儲目錄 dataLogDir=/datalog # 集群模式下 節(jié)點之間的心跳時間2s（每2s進行一次心跳檢測） tickTime=2000 # 集群初始化時 節(jié)點之間同步次數(shù)。（5 * 2s = 10s 10s內(nèi)未初始化成功則初始化失?。? initLimit=5 # 集群模式下 同步數(shù)據(jù)次數(shù)。 （2 * 2s = 4s 4s內(nèi)未同步則超時失?。? syncLimit=2 # 數(shù)據(jù)快照保留個數(shù) autopurge.snapRetainCount=3 # 數(shù)據(jù)快照清除時間間隔（單位為小時） 0：不清除 ，1：1小時 autopurge.purgeInterval=0 # 最大的客戶端鏈接數(shù) maxClientCnxns=60
4. 基礎(chǔ)命令
不同版本之前的命令會有所差異，本章是基于zk:3.7版本。官方地址
4.1 創(chuàng)建會話
首先執(zhí)行命令，創(chuàng)建新的會話，進入終端。
# 進入到zk的安裝目錄的bin目錄下執(zhí)行 $ ./zkCli.sh # 如果zk不是在本機 則可以使用server參數(shù)指定鏈接地址 $ ./zkCli.sh -server 127.0.0.1:2181
4.2 ls
ls [-s] [-w] [-R] path：ls 命令用于查看某個路徑下目錄列表。
-s：查看此znode狀態(tài)信息。-w：監(jiān)聽此znode目錄變化。-R：遞歸查看
$ ls / # 查看根目錄 [app, app10000000000, zookeeper] $ ls /app [] $ ls -s /app [app01, app02] cZxid = 0x1b ctime = Sun Aug 29 13:07:24 UTC 2021 mZxid = 0x1b mtime = Sun Aug 29 13:07:24 UTC 2021 pZxid = 0x2e cversion = 2 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 0 numChildren = 2
監(jiān)聽znode目錄變化：
4.3 create
create [-s] [-e] [-c] [-t ttl] path [data] [acl]：create 用于創(chuàng)建znode。
-s：創(chuàng)建順序節(jié)點。-e：創(chuàng)建臨時節(jié)點。-c：創(chuàng)建一個容器節(jié)點，當容器中的最后一個子節(jié)點被刪除時，容器也隨之消失。-t：設(shè)置節(jié)點存活時間（毫秒）。
⚠️注意：如果要設(shè)置超時時間需在配置文件中激活配置：
zookeeper.extendedTypesEnabled=true
data：設(shè)置的數(shù)據(jù)。acl：訪問控制配置。
$ ls / [app, app10000000000, zookeeper] # 創(chuàng)建持久節(jié)點 /app3 $ create /app3 zhangtieniu Created /app3 # 創(chuàng)建有序的持久節(jié)點 /app4 $ create -s /app4 zhangsan Created /app40000000011 # 創(chuàng)建臨時節(jié)點 /app5 $ create -e /app5 linshi Created /app5 # 創(chuàng)建臨時有序節(jié)點 /app6 $ create -s -e /app6 linshiyouxu Created /app60000000013 $ ls / [app, app10000000000, app3, app40000000011, app5, app60000000013, zookeeper]
4.4 get
get [-s] [-w] path： 命令用于獲取節(jié)點數(shù)據(jù)和狀態(tài)信息。
-s：查看此znode狀態(tài)信息。-w：監(jiān)聽此znode值變化。
$ get /app3 # 查看/app3 znode內(nèi)容 zhangtieniu
監(jiān)聽znod值變化：
4.5 stat
stat [-w] path：命令用于查看節(jié)點狀態(tài)信息。
-w：監(jiān)聽節(jié)點狀態(tài)變化。
$ stat /app3 cZxid = 0x34 # 創(chuàng)建節(jié)點時的事務(wù)id ctime = Sun Aug 29 14:31:52 UTC 2021 # 創(chuàng)建時間 mZxid = 0x34 # 創(chuàng)建的版本號 mtime = Sun Aug 29 14:31:52 UTC 2021 # 修改時間 pZxid = 0x34 # 子節(jié)點列表最后一次被修改的事務(wù)id cversion = 0 # 節(jié)點版本號 dataVersion = 0 # 數(shù)據(jù)版本號 aclVersion = 0 # 權(quán)限版本號 ephemeralOwner = 0x0 # 臨時擁有者 dataLength = 11 # 節(jié)點值數(shù)據(jù)長度 numChildren = 0 # 子節(jié)點數(shù)量
4.6 set
set [-s] [-v version] path data： 命令用于修改節(jié)點存儲的數(shù)據(jù)。
-s：查看設(shè)置成功后的狀態(tài)信息。
-v：指定設(shè)置值的版本號，該值只能為該節(jié)點的最新版本。可以使用其實現(xiàn)樂觀鎖。
$ set /app wangmazi $ set -s /app lisi cZxid = 0x1b ctime = Sun Aug 29 13:07:24 UTC 2021 mZxid = 0x43 mtime = Sun Aug 29 14:50:52 UTC 2021 pZxid = 0x3f cversion = 4 dataVersion = 6 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 4 numChildren = 4
4.7 delete
delete [-v version] path：刪除指定節(jié)點。
⚠️注意：當刪除的節(jié)點有子節(jié)點時，不能使用該命令。需使用deleteall命令刪除。
-v：刪除指定版本。與set同理
$ delete /app3
4.8 quit
關(guān)閉會話。
5. 節(jié)點的監(jiān)聽機制
zk的監(jiān)聽機制，可以使客戶端可以監(jiān)聽znode節(jié)點的變化，znode節(jié)點的變化出發(fā)相應(yīng)的事件，然后清除對該節(jié)點的檢測。
⚠️注意：在這里再強調(diào)一次，當使用zkCli.sh 創(chuàng)建會話時，對znode的監(jiān)聽只能觸發(fā)一次。但在使用java客戶端鏈接時，可以一直觸發(fā)。
$ ls -w /path # 監(jiān)聽節(jié)點目錄的變化 $ get -w /path # 監(jiān)聽節(jié)點數(shù)據(jù)的變化
6. quick start
6.1 項目結(jié)構(gòu)
.
├── pom.xml
└── src
└── main
├── java
│ └── com
│ └── ldx
│ └── zookeeper
│ ├── ZookeeperApplication.java # 啟動類
│ ├── config
│ │ ├── CuratorClientConfig.java # zk配置類
│ │ └── CuratorClientProperties.java # zk 配置屬性文件
│ ├── controller
│ │ └── AppController.java # zk 測試文件
│ └── util
│ └── CuratorClient.java # zk工具類
└── resources
└── application.yaml # 服務(wù)配置文件
6.2 引入依賴
Curator 是 Netflix 公司開源的一套 zookeeper 客戶端框架，解決了很多 Zookeeper 客戶端非常底層的細節(jié)開發(fā)工作，包括連接重連、反復(fù)注冊 Watcher 和 NodeExistsException 異常等。
curator-recipes：封裝了一些高級特性，如：Cache 事件監(jiān)聽、選舉、分布式鎖、分布式計數(shù)器、分布式 Barrier 等。
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.5.4</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.ldx</groupId>
<artifactId>zookeeper</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>zookeeper</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- client 操作工具包 -->
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>5.2.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<excludes>
<exclude>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>
</project>
6.3 application.
server:
port: 8080
# curator配置
curator-client:
# 連接字符串
connection-string: localhost:2181
# 根節(jié)點
namespace: ldx
# 節(jié)點數(shù)據(jù)編碼
charset: utf8
# session超時時間
session-timeout-ms: 60000
# 連接超時時間
connection-timeout-ms: 15000
# 關(guān)閉連接超時時間
max-close-wait-ms: 1000
# 默認數(shù)據(jù)
default-data: ""
# 當半數(shù)以上zookeeper服務(wù)出現(xiàn)故障仍然提供讀服務(wù)
can-be-read-only: false
# 自動創(chuàng)建父節(jié)點
use-container-parents-if-available: true
# 重試策略，默認使用BoundedExponentialBackoffRetry
retry:
max-sleep-time-ms: 10000
base-sleep-time-ms: 1000
max-retries: 3
# 認證信息
#auth:
#scheme: digest
# auth: username:password
6.4 CuratorClientProperties
package com.ldx.zookeeper.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* zk 屬性配置類
*
* @author ludangxin
* @date 2021/8/31
*/
@Data
@ConfigurationProperties(prefix = "curator-client")
public class CuratorClientProperties {
/**
* 連接地址
*/
private String connectionString;
/**
* 命名空間
*/
private String namespace;
/**
* 字符集
*/
private String charset = "utf8";
/**
* 會話超時時間 毫秒
*/
private int sessionTimeoutMs = 60000;
/**
* 連接超時時間 毫秒
*/
private int connectionTimeoutMs = 15000;
/**
* 最大關(guān)閉等待時間 毫秒
*/
private int maxCloseWaitMs = 1000;
/**
* 默認數(shù)據(jù)
*/
private String defaultData = "";
/**
* 當半數(shù)以上zookeeper服務(wù)出現(xiàn)故障仍然提供讀服務(wù)
*/
private boolean canBeReadOnly = false;
/**
* 自動創(chuàng)建父節(jié)點
*/
private boolean useContainerParentsIfAvailable = true;
/**
* 線程池名稱
*/
private String threadFactoryClassName;
private Retry retry = new Retry();
private Auth auth = new Auth();
@Data
public static class Retry {
private int maxSleepTimeMs = 10000;
private int baseSleepTimeMs = 1000;
private int maxRetries = 3;
}
@Data
public static class Auth {
private String scheme = "digest";
private String auth;
}
}
6.5 CuratorClientConfig
package com.ldx.zookeeper.config;
import com.ldx.zookeeper.util.CuratorClient;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.ensemble.EnsembleProvider;
import org.apache.curator.ensemble.fixed.FixedEnsembleProvider;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.api.CompressionProvider;
import org.apache.curator.framework.imps.GzipCompressionProvider;
import org.apache.curator.retry.BoundedExponentialBackoffRetry;
import org.apache.curator.utils.DefaultZookeeperFactory;
import org.apache.curator.utils.ZookeeperFactory;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.ThreadFactory;
/**
* zk 配置類
*
* @author ludangxin
* @date 2021/8/31
*/
@Slf4j
@Configuration
@EnableConfigurationProperties(CuratorClientProperties.class)
public class CuratorClientConfig {
@Bean
public EnsembleProvider ensembleProvider(CuratorClientProperties curatorClientProperties) {
return new FixedEnsembleProvider(curatorClientProperties.getConnectionString());
}
@Bean
public RetryPolicy retryPolicy(CuratorClientProperties curatorClientProperties) {
CuratorClientProperties.Retry retry = curatorClientProperties.getRetry();
return new BoundedExponentialBackoffRetry(retry.getBaseSleepTimeMs(), retry.getMaxSleepTimeMs(), retry.getMaxRetries());
}
@Bean
public CompressionProvider compressionProvider() {
return new GzipCompressionProvider();
}
@Bean
public ZookeeperFactory zookeeperFactory() {
return new DefaultZookeeperFactory();
}
@Bean
public ACLProvider aclProvider() {
return new ACLProvider() {
@Override
public List<ACL> getDefaultAcl() {
return ZooDefs.Ids.CREATOR_ALL_ACL;
}
@Override
public List<ACL> getAclForPath(String path) {
return ZooDefs.Ids.CREATOR_ALL_ACL;
}
};
}
@Bean
@SneakyThrows
public CuratorFrameworkFactory.Builder builder(EnsembleProvider ensembleProvider,
RetryPolicy retryPolicy,
CompressionProvider compressionProvider,
ZookeeperFactory zookeeperFactory,
ACLProvider aclProvider,
CuratorClientProperties curatorClientProperties) {
String charset = curatorClientProperties.getCharset();
CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
.ensembleProvider(ensembleProvider)
.retryPolicy(retryPolicy)
.compressionProvider(compressionProvider)
.zookeeperFactory(zookeeperFactory)
.namespace(curatorClientProperties.getNamespace())
.sessionTimeoutMs(curatorClientProperties.getSessionTimeoutMs())
.connectionTimeoutMs(curatorClientProperties.getConnectionTimeoutMs())
.maxCloseWaitMs(curatorClientProperties.getMaxCloseWaitMs())
.defaultData(curatorClientProperties.getDefaultData().getBytes(Charset.forName(charset)))
.canBeReadOnly(curatorClientProperties.isCanBeReadOnly());
if (!curatorClientProperties.isUseContainerParentsIfAvailable()) {
builder.dontUseContainerParents();
}
CuratorClientProperties.Auth auth = curatorClientProperties.getAuth();
if (StringUtils.isNotBlank(auth.getAuth())) {
builder.authorization(auth.getScheme(), auth.getAuth().getBytes(Charset.forName(charset)));
builder.aclProvider(aclProvider);
}
String threadFactoryClassName = curatorClientProperties.getThreadFactoryClassName();
if (StringUtils.isNotBlank(threadFactoryClassName)) {
try {
Class<?> cls = Class.forName(threadFactoryClassName);
ThreadFactory threadFactory = (ThreadFactory) cls.newInstance();
builder.threadFactory(threadFactory);
} catch (Exception e) {
log.error("init CuratorClient error", e);
}
}
return builder;
}
@Bean(initMethod = "init", destroyMethod = "stop")
public CuratorClient curatorClient(CuratorFrameworkFactory.Builder builder) {
return new CuratorClient(builder);
}
}
6.6 CuratorClient
package com.ldx.zookeeper.util;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.locks.*;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.zookeeper.CreateMode;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
/**
* zookeeper工具類
*
* @author ludangxin
* @date 2021/8/30
*/
@Slf4j
public class CuratorClient {
/**
* 默認的字符編碼集
*/
private static final String DEFAULT_CHARSET = "utf8";
/**
* 客戶端
*/
private final CuratorFramework client;
/**
* 字符集
*/
private String charset = DEFAULT_CHARSET;
@SneakyThrows
public CuratorClient(CuratorFrameworkFactory.Builder builder) {
client = builder.build();
}
@SneakyThrows
public CuratorClient(CuratorFrameworkFactory.Builder builder, String charset) {
client = builder.build();
this.charset = charset;
}
public void init() {
client.start();
client.getConnectionStateListenable().addListener((client, state) -> {
if (state==ConnectionState.LOST) {
// 連接丟失
log.info("lost session with zookeeper");
} else if (state==ConnectionState.CONNECTED) {
// 連接新建
log.info("connected with zookeeper");
} else if (state==ConnectionState.RECONNECTED) {
// 重新連接
log.info("reconnected with zookeeper");
}
});
}
/**
* 關(guān)閉會話
*/
public void stop() {
log.info("zookeeper session close");
client.close();
}
/**
* 創(chuàng)建節(jié)點
*
* @param mode 節(jié)點類型
* 1、PERSISTENT 持久化目錄節(jié)點，存儲的數(shù)據(jù)不會丟失。
* 2、PERSISTENT_SEQUENTIAL順序自動編號的持久化目錄節(jié)點，存儲的數(shù)據(jù)不會丟失
* 3、EPHEMERAL臨時目錄節(jié)點，一旦創(chuàng)建這個節(jié)點的客戶端與服務(wù)器端口也就是session 超時，這種節(jié)點會被自動刪除
* 4、EPHEMERAL_SEQUENTIAL臨時自動編號節(jié)點，一旦創(chuàng)建這個節(jié)點的客戶端與服務(wù)器端口也就是session 超時，這種節(jié)點會被自動刪除，并且根據(jù)當前已經(jīng)存在的節(jié)點數(shù)自動加 1，然后返回給客戶端已經(jīng)成功創(chuàng)建的目錄節(jié)點名。
* @param path 節(jié)點名稱
* @param nodeData 節(jié)點數(shù)據(jù)
*/
@SneakyThrows
public void createNode(CreateMode mode, String path, String nodeData) {
// 使用creatingParentContainersIfNeeded()之后Curator能夠自動遞歸創(chuàng)建所有所需的父節(jié)點
client.create().creatingParentsIfNeeded().withMode(mode).forPath(path, nodeData.getBytes(Charset.forName(charset)));
}
/**
* 創(chuàng)建節(jié)點
*
* @param mode 節(jié)點類型
* 1、PERSISTENT 持久化目錄節(jié)點，存儲的數(shù)據(jù)不會丟失。
* 2、PERSISTENT_SEQUENTIAL順序自動編號的持久化目錄節(jié)點，存儲的數(shù)據(jù)不會丟失
* 3、EPHEMERAL臨時目錄節(jié)點，一旦創(chuàng)建這個節(jié)點的客戶端與服務(wù)器端口也就是session 超時，這種節(jié)點會被自動刪除
* 4、EPHEMERAL_SEQUENTIAL臨時自動編號節(jié)點，一旦創(chuàng)建這個節(jié)點的客戶端與服務(wù)器端口也就是session 超時，這種節(jié)點會被自動刪除，并且根據(jù)當前已經(jīng)存在的節(jié)點數(shù)自動加 1，然后返回給客戶端已經(jīng)成功創(chuàng)建的目錄節(jié)點名。
* @param path 節(jié)點名稱
*/
@SneakyThrows
public void createNode(CreateMode mode, String path) {
// 使用creatingParentContainersIfNeeded()之后Curator能夠自動遞歸創(chuàng)建所有所需的父節(jié)點
client.create().creatingParentsIfNeeded().withMode(mode).forPath(path);
}
/**
* 刪除節(jié)點數(shù)據(jù)
*
* @param path 節(jié)點名稱
*/
@SneakyThrows
public void deleteNode(final String path) {
deleteNode(path, true);
}
/**
* 刪除節(jié)點數(shù)據(jù)
*
* @param path 節(jié)點名稱
* @param deleteChildre 是否刪除子節(jié)點
*/
@SneakyThrows
public void deleteNode(final String path, Boolean deleteChildre) {
if (deleteChildre) {
// guaranteed()刪除一個節(jié)點，強制保證刪除,
// 只要客戶端會話有效，那么Curator會在后臺持續(xù)進行刪除操作，直到刪除節(jié)點成功
client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
} else {
client.delete().guaranteed().forPath(path);
}
}
/**
* 設(shè)置指定節(jié)點的數(shù)據(jù)
*
* @param path 節(jié)點名稱
* @param data 節(jié)點數(shù)據(jù)
*/
@SneakyThrows
public void setNodeData(String path, String data) {
client.setData().forPath(path, data.getBytes(Charset.forName(charset)));
}
/**
* 獲取指定節(jié)點的數(shù)據(jù)
*
* @param path 節(jié)點名稱
* @return 節(jié)點數(shù)據(jù)
*/
@SneakyThrows
public String getNodeData(String path) {
return new String(client.getData().forPath(path), Charset.forName(charset));
}
/**
* 獲取數(shù)據(jù)時先同步
*
* @param path 節(jié)點名稱
* @return 節(jié)點數(shù)據(jù)
*/
public String synNodeData(String path) {
client.sync();
return getNodeData(path);
}
/**
* 判斷節(jié)點是否存在
*
* @param path 節(jié)點名稱
* @return true 節(jié)點存在，false 節(jié)點不存在
*/
@SneakyThrows
public boolean isExistNode(final String path) {
client.sync();
return Objects.nonNull(client.checkExists().forPath(path));
}
/**
* 獲取節(jié)點的子節(jié)點
*
* @param path 節(jié)點名稱
* @return 子節(jié)點集合
*/
@SneakyThrows
public List<String> getChildren(String path) {
return client.getChildren().forPath(path);
}
/**
* 創(chuàng)建排他鎖
*
* @param path 節(jié)點名稱
* @return 排他鎖
*/
public InterProcessSemaphoreMutex getSemaphoreMutexLock(String path) {
return new InterProcessSemaphoreMutex(client, path);
}
/**
* 創(chuàng)建可重入排他鎖
*
* @param path 節(jié)點名稱
* @return 可重入排他鎖
*/
public InterProcessMutex getMutexLock(String path) {
return new InterProcessMutex(client, path);
}
/**
* 創(chuàng)建一組可重入排他鎖
*
* @param paths 節(jié)點名稱集合
* @return 鎖容器
*/
public InterProcessMultiLock getMultiMutexLock(List<String> paths) {
return new InterProcessMultiLock(client, paths);
}
/**
* 創(chuàng)建一組任意類型的鎖
*
* @param locks 鎖集合
* @return 鎖容器
*/
public InterProcessMultiLock getMultiLock(List<InterProcessLock> locks) {
return new InterProcessMultiLock(locks);
}
/**
* 加鎖
*
* @param lock 分布式鎖對象
*/
@SneakyThrows
public void acquire(InterProcessLock lock) {
lock.acquire();
}
/**
* 加鎖
*
* @param lock 分布式鎖對象
* @param time 等待時間
* @param unit 時間單位
*/
@SneakyThrows
public void acquire(InterProcessLock lock, long time, TimeUnit unit) {
lock.acquire(time, unit);
}
/**
* 釋放鎖
*
* @param lock 分布式鎖對象
*/
@SneakyThrows
public void release(InterProcessLock lock) {
lock.release();
}
/**
* 檢查是否當前jvm的線程獲取了鎖
*
* @param lock 分布式鎖對象
* @return true/false
*/
public boolean isAcquiredInThisProcess(InterProcessLock lock) {
return lock.isAcquiredInThisProcess();
}
/**
* 獲取讀寫鎖
*
* @param path 節(jié)點名稱
* @return 讀寫鎖
*/
public InterProcessReadWriteLock getReadWriteLock(String path) {
return new InterProcessReadWriteLock(client, path);
}
/**
* 監(jiān)聽數(shù)據(jù)節(jié)點的變化情況
*
* @param path 節(jié)點名稱
* @param listener 監(jiān)聽器
* @return 監(jiān)聽節(jié)點的TreeCache實例
*/
@SneakyThrows
public CuratorCache watch(String path, CuratorCacheListener listener) {
CuratorCache curatorCache = CuratorCache.builder(client, path).build();
curatorCache.listenable().addListener(listener);
curatorCache.start();
return curatorCache;
}
/**
* 監(jiān)聽數(shù)據(jù)節(jié)點的變化情況
*
* @param path 節(jié)點名稱
* @param listener 監(jiān)聽器
* @return 監(jiān)聽節(jié)點的TreeCache實例
*/
public CuratorCache watch(String path, CuratorCacheListener listener, Executor executor) {
CuratorCache curatorCache = CuratorCache.builder(client, path).build();
curatorCache.listenable().addListener(listener, executor);
curatorCache.start();
return curatorCache;
}
/**
* 取消監(jiān)聽節(jié)點
*
* @param path 節(jié)點名稱
* @param listener 監(jiān)聽器
*/
public void unwatch(String path, CuratorCacheListener listener) {
CuratorCache curatorCache = CuratorCache.builder(client, path).build();
curatorCache.listenable().removeListener(listener);
}
}
6.7 AppController
package com.ldx.zookeeper.controller;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ldx.zookeeper.util.CuratorClient;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.web.bind.annotation.*;
/**
* demo
*
* @author ludangxin
* @date 2021/8/30
*/
@Slf4j
@RestController
@RequestMapping("app")
@RequiredArgsConstructor
public class AppController {
private final CuratorClient curatorClient;
@GetMapping("{appName}")
public String getData(@PathVariable String appName) {
return curatorClient.getNodeData(setPrefix(appName));
}
@PostMapping("{appName}")
public String addApp(@PathVariable String appName, @RequestParam String data) {
curatorClient.createNode(CreateMode.PERSISTENT, setPrefix(appName), data);
return "ok";
}
@PostMapping("{appName}/{childName}")
public String addAppChild(@PathVariable String appName, @PathVariable String childName, @RequestParam String data) {
curatorClient.createNode(CreateMode.PERSISTENT, setPrefix(appName).concat(setPrefix(childName)), data);
return "ok";
}
@PutMapping("{appName}")
public String setData(@PathVariable String appName, String data) {
curatorClient.setNodeData(setPrefix(appName), data);
return "ok";
}
@PutMapping("{appName}/{childName}")
public String setData(@PathVariable String appName, @PathVariable String childName, String data) {
curatorClient.setNodeData(setPrefix(appName).concat(setPrefix(childName)), data);
return "ok";
}
@DeleteMapping("{appName}")
public String delApp(@PathVariable String appName) {
curatorClient.deleteNode(setPrefix(appName));
return "ok";
}
@PostMapping("{appName}/watch/dir")
public String watchAppDir(@PathVariable String appName) {
CuratorCacheListener listener = CuratorCacheListener.builder().forDeletes(obj -> {
String path = obj.getPath();
String data = new String(obj.getData());
Stat stat = obj.getStat();
log.info("節(jié)點：{} 被刪除，節(jié)點數(shù)據(jù)：{}，節(jié)點狀態(tài)：\\{version：{}，createTime:{}\\}", path, data,
stat.getVersion(), stat.getCtime());
}).build();
curatorClient.watch(setPrefix(appName), listener);
return "ok";
}
@PostMapping("{appName}/watch/data")
public String watchAppData(@PathVariable String appName) {
ObjectMapper mapper = new ObjectMapper();
CuratorCacheListener listener = CuratorCacheListener.builder().forChanges((oldNode, newNode) -> {
try {
String path = oldNode.getPath();
log.info("節(jié)點：{} 被修改，修改前：{} ", path, mapper.writeValueAsString(oldNode));
log.info("節(jié)點：{} 被修改，修改后：{} ", path, mapper.writeValueAsString(newNode));
} catch(JsonProcessingException e) {
e.printStackTrace();
}
}).build();
curatorClient.watch(setPrefix(appName), listener);
return "ok";
}
private String setPrefix(String appName) {
String prefix = "/";
if(!appName.startsWith(prefix)) {
appName = prefix.concat(appName);
}
return appName;
}
}
7. ZAB 協(xié)議介紹
ZAB（ZooKeeper Atomic Broadcast 原子廣播） 協(xié)議是為分布式協(xié)調(diào)服務(wù) ZooKeeper 專門設(shè)計的一種支持崩潰恢復(fù)的原子廣播協(xié)議。 在 ZooKeeper 中，主要依賴 ZAB 協(xié)議來實現(xiàn)分布式數(shù)據(jù)一致性，基于該協(xié)議，ZooKeeper 實現(xiàn)了一種主備模式的系統(tǒng)架構(gòu)來保持集群中各個副本之間的數(shù)據(jù)一致性。
ZAB 協(xié)議兩種基本的模式：崩潰恢復(fù)和消息廣播
ZAB協(xié)議包括兩種基本的模式，分別是 崩潰恢復(fù)和消息廣播。當整個服務(wù)框架在啟動過程中，或是當 Leader 服務(wù)器出現(xiàn)網(wǎng)絡(luò)中斷、崩潰退出與重啟等異常情況時，ZAB 協(xié)議就會進人恢復(fù)模式并選舉產(chǎn)生新的Leader服務(wù)器。當選舉產(chǎn)生了新的 Leader 服務(wù)器，同時集群中已經(jīng)有過半的機器與該Leader服務(wù)器完成了狀態(tài)同步之后，ZAB協(xié)議就會退出恢復(fù)模式。其中，所謂的狀態(tài)同步是指數(shù)據(jù)同步，用來保證集群中存在過半的機器能夠和Leader服務(wù)器的數(shù)據(jù)狀態(tài)保持一致。
當集群中已經(jīng)有過半的Follower服務(wù)器完成了和Leader服務(wù)器的狀態(tài)同步，那么整個服務(wù)框架就可以進人消息廣播模式了。 當一臺同樣遵守ZAB協(xié)議的服務(wù)器啟動后加人到集群中時，如果此時集群中已經(jīng)存在一個Leader服務(wù)器在負責(zé)進行消息廣播，那么新加人的服務(wù)器就會自覺地進人數(shù)據(jù)恢復(fù)模式：找到Leader所在的服務(wù)器，并與其進行數(shù)據(jù)同步，然后一起參與到消息廣播流程中去。正如上文介紹中所說的，ZooKeeper設(shè)計成只允許唯一的一個Leader服務(wù)器來進行事務(wù)請求的處理。Leader服務(wù)器在接收到客戶端的事務(wù)請求后，會生成對應(yīng)的事務(wù)提案并發(fā)起一輪廣播協(xié)議；而如果集群中的其他機器接收到客戶端的事務(wù)請求，那么這些非Leader服務(wù)器會首先將這個事務(wù)請求轉(zhuǎn)發(fā)給Leader服務(wù)器。
以上就是詳解Zookeeper基礎(chǔ)知識的詳細內(nèi)容，更多關(guān)于Zookeeper基礎(chǔ)的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
下面小編就為大家?guī)硪黄狫ava基礎(chǔ)的幾道練習(xí)題(分享)。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧，希望可以幫到你2021-07-07
java中@DateTimeFormat和@JsonFormat注解的使用
本文主要介紹了java中@DateTimeFormat和@JsonFormat注解的使用，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-08-08
SpringBoot之logback-spring.xml不生效的解決方法
這篇文章主要介紹了SpringBoot之logback-spring.xml不生效的解決方法，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-01-01
這篇文章主要介紹了Java8排序stream.sorted()的使用，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03
Java簡單統(tǒng)計字符串中漢字，英文字母及數(shù)字數(shù)量的方法
這篇文章主要介紹了Java簡單統(tǒng)計字符串中漢字，英文字母及數(shù)字數(shù)量的方法,涉及java針對字符串的遍歷、編碼轉(zhuǎn)換、判斷等相關(guān)操作技巧,需要的朋友可以參考下2017-06-06
本篇文章主要介紹了Spring循環(huán)依賴的三種方式，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-01-01
這篇文章主要介紹了解讀jdk動態(tài)代理為什么必須實現(xiàn)接口問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-02-02
下面小編就為大家?guī)硪黄獪\析Java編程中類和對象的定義。小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧，祝大家游戲愉快哦2016-05-05
這篇文章主要給大家介紹了關(guān)于Spring解密之XML解析與Bean注冊的相關(guān)資料，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面小編來一起學(xué)習(xí)學(xué)習(xí)吧。2018-01-01
idea輸入sout無法自動補全System.out.println()的問題
這篇文章主要介紹了idea輸入sout無法自動補全System.out.println()的問題,本文給大家分享解決方案，供大家參考，需要的朋友可以參考下2020-07-07