AI 助理

你好，我是AI助理，可以解答问题、推荐解决方案等

开发者社区 数据库 文章 正文

【SpringCloud Alibaba系列】一文全面解析Zookeeper安装、常用命令、JavaAPI操作、Watch事件监听、分布式锁、集群搭建、核心理论

2024-12-24 501
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
蓝染-惣右介
简介： 一文全面解析Zookeeper安装、常用命令、JavaAPI操作、Watch事件监听、分布式锁、集群搭建、核心理论。

Zookeeper


学习目标

  • Zookeeper简介
  • ZooKeeper安装与配置
  • ZooKeeper命令操作
  • ZooKeeperJavaAPI操作
  • ZooKeeper集群搭建
  • Zookeeper核心理论

一、简介

Zookeeper是 Apache Hadoop项目下的一个子项目，是一个树形目录服务。

Zookeeper翻译过来就是动物园管理员，他是用来管Hadoop(大象)、Hive(蜜蜂)、Pig(小猪)的管理员。简称ZK。

Zookeeper是一个分布式的、开源的分布式应用程序的协调服务。

官网：https://zookeeper.apache.org/

二、应用场景

Zookeeper是一个经典的分布式数据一致性解决方案，致力于为分布式应用提供一个高性能、高可用，且具有严格顺序访问控制能力的分布式协调存储服务。

Zookeeper提供的主要功能包括：

  • 作为配置中心维护配置信息
  • 提供分布式锁服务
  • 作为注册中心实现集群管理
  • 生成分布式唯一ID
  1. 配置管理

java编程经常会遇到配置项，比如数据库的url、 schema、user和 password等。通常这些配置项我们会放置在配置文件中，再将配置文件放置在服务器上当需要更改配置项时，需要去服务器上修改对应的配置文件。

但是随着分布式系统的兴起,由于许多服务都需要使用到该配置文件,因此有必须保证该配置服务的高可用性(highavailability)和各台服务器上配置数据的一致性。

通常会将配置文件部署在一个集群上，然而一个集群动辄上千台服务器，此时如果再一台台服务器逐个修改配置文件那将是非常繁琐且危险的的操作，因此就需要一种服务，能够高效快速且可靠地完成配置项的更改等操作，并能够保证各配置项在每台服务器上的数据一致性。

zookeeper就可以提供这样一种服务，其使用Zab这种一致性协议来保证一致性。现在有很多开源项目使用zookeeper来维护配置，如在 hbase中，客户端就是连接一个 zookeeper，获得必要的 hbase集群的配置信息，然后才可以进一步操作。还有在开源的消息队列 kafka中，也便用zookeeper来维护 brokers的信息。在 alibaba开源的soa框架dubbo中也广泛的使用zookeeper管理一些配置来实现服务治理。

  1. 分布式锁

一个集群是一个分布式系统，由多台服务器组成。为了提高并发度和可靠性，多台服务器上运行着同一种服务。当多个服务在运行时就需要协调各服务的进度，有时候需要保证当某个服务在进行某个操作时，其他的服务都不能进行该操作，即对该操作进行加锁，如果当前机器挂掉后，释放锁并 fail over到其他的机器继续执行该服务。

  1. 集群管理

一个集群有时会因为各种软硬件故障或者网络故障，出现棊些服务器挂掉而被移除集群，而某些服务器加入到集群中的情况，zookeeper会将这些服务器加入/移出的情况通知给集群中的其他正常工作的服务器，以及时调整存储和计算等任务的分配和执行等。此外zookeeper还会对故障的服务器做出诊断并尝试修复。

  1. 生成分布式唯一ID

在过去的单库单表型系统中，通常可以使用数据库字段自带的auto_increment属性来自动为每条记录生成一个唯一的ID。

但是分库分表后，就无法在依靠数据库的auto_increment属性来唯一标识一条记录了。此时我们就可以用Zookeeper在分布式环境下生成全局唯一ID。

做法如下：每次要生成一个新id时，创建一个持久顺序节点，创建操作返回的节点序号，即为新id，然后把比自己节点小的删除即可。

三、设计目标

Zookeeper致力于为分布式应用提供一个高性能、高可用，且具有严格顺序访问控制能力的分布式协调服务。

  1. 高性能

Zookeeper将全量数据存储在内存中，并直接服务于客户端的所有非事务请求，尤其用于以读为主的应用场景。

  1. 高可用

Zookeeper一般以集群的方式对外提供服务，一般3~5台机器就可以组成一个可用的 Zookeeper集群了，每台机器都会在内存中维护当前的服务器状态，井且每台机器之间都相互保持着通信。只要集群中超过一半的机器都能够正常工作，那么整个集群就能够正常对外服务。

  1. 严格顺序访问

对于来自客户端的每个更新请求，Zookeeper都会分配一个全局唯一的递增编号，这个编号反应了所有事务操作的先后顺序。

四、数据模型

ZooKeeper是一个树形目录服务，其数据模型和Unix的文件系统目录树很类似，拥有一个层次化结构。这里面的每一个节点都被称为：ZNode，每个节点上都会保存自己的数据和节点信息。

节点可以拥有子节点，同时也允许少量(1MB）数据存储在该节点之下。

节点可以分为四大类:

  • PERSISTENT持久化节点
  • EPHEMERAL 临时节点：-e
  • PERSISTENT_SEQUENTIAL持久化顺序节点：-s
  • EPHEMERAL_SEQUENTIAL临时顺序节点：-es

那么如何描述一个znode呢？一个znode大体上分为3个部分：

  • 结点的数据：即znode data(结点path，结点data)的关系就像是Java map中的 key value关系。
  • 结点的子结点children。
  • 结点的状态stat：用来描述当前结点的创建、修改记录，包括cZxid、ctime等。

结点类型

zookeeper中的结点有两种，分别为临时结点和永久结点。结点的类型在创建时被确定，并且不能改变

  • 临时节点：
  • 该节点的生命周期依赖于创建它们的会话。一旦会话( Session）结束，临时节点将被自动删除，当然可以也可以手动删除。虽然每个临时的 Znode都会绑定到一个客户端会话，但他们对所有的客户端还是可见的。另外，Zookeeper的临时节点不允许拥有子节点。
  • 持久化结点：
  • 该结点的生命周期不依赖于会话，并且只有在客户端显示执行删除操作的时候，它们才能被删除。

五、单机安装

  • 测试系统环境CentOS7.9
  • zookeeper：apache-zookeeper-3.5.6-bin.tar.gz
  • jdk（v1.8及以上）：jdk-8u171-linux-x64.tar.gz

下载地址：http://archive.apache.org/dist/zookeeper/

  1. zookeeper底层依赖于jdk，它运行在JVM之上。根目录下先进行jdk的安装（已安装可跳过），jdk使用 jdk-8u171-linux-x64.tar.gz
tar -zxvf jdk-8u171-linux-x64.tar.gz
  1. 配置jdk环境变量。
vim /etc/profile JAVA_HOME=/usr/local/jdk1.8.0_171 export JAVA_HOME PATH=$JAVA_HOME/bin:$PATH export PATH source /etc/profile
  1. 检测jdk安装java -version，如果反馈了Java信息，则成功。
  2. 将下载的ZooKeeper放到/opt/zookeeper目录下。
# 上传zookeeper安装包到服务器 put f:/setup/apache-zookeeper-3.5.6-bin.tar.gz # 在/opt目录下创建zookeeper目录 mkdir /opt/zookeeper # 将zookeeper安装包移动到 /opt/zookeeper mv apache-zookeeper-3.5.6-bin.tar.gz /opt/zookeeper/
  1. 将tar包解压到/opt/zookeeper目录下。
tar -zxvf apache-zookeeper-3.5.6-bin.tar.gz -C /opt/zookeeper/
  1. 为zookeeper准备配置文件zoo.cfg，进入到conf目录拷贝一个zoo_sample.cfg并完成配置，修改zoo.cfg。
# 进入到conf目录 cd /opt/zookeeper/apache-zookeeper-3.5.6-bin/conf/ # 拷贝 (配置文件必须叫zoo.cfg才能生效) cp zoo_sample.cfg zoo.cfg # 打开目录 cd /opt/zookeeper/ # 创建zookeeper存储目录 mkdir zkdata # 修改zoo.cfg vim /opt/zookeeper/apache-zookeeper-3.5.6-bin/conf/zoo.cfg # 修改存储目录 将zk默认存储数据的临时目录dataDir=/tmp/zookeeper，修改为：dataDir=/opt/zookeeper/zkdata

  1. 2181是ZK监听客户端连接的默认端口号，记得防火墙把这个端口给开放出来，不然客户端连接不上。
# 添加2181端口到防火墙允许列表 firewall-cmd --zone=public --add-port=2181/tcp --permanent # 立即生效 firewall-cmd --reload # 查看Linux防火墙允许通过的端口号 firewall-cmd --zone=public --list-ports
  1. 启动zookeeper
# 进入zookeeper的bin目录 cd /opt/zookeeper/apache-zookeeper-3.5.6-bin/bin/ # 启动zookeeper ./zkServer.sh start # 查看ZooKeeper状态 #./zkServer.sh status # 停止ZooKeeper #./zkServer.sh stop # 开启ZooKeeper #./zkServer.sh start # 重启ZooKeeper #./zkServer.sh restart # 进入zookeeper客户端内部 ./zkCli.sh

zookeeper单节点standalone启动成功。

六、命令操作

1. 服务端常用命令

  • 启动ZooKeeper服务：./zkServer.sh start
  • 查看ZooKeeper服务状态：./zkServer.sh status
  • 停止ZooKeeper服务：./zkServer.sh stop
  • 重启ZooKeeper服务：./zkServer.sh restart

2. 客户端常用命令

  1. ./zkCli.sh -server ip:portZookeeper客户端连接服务端。
# Zookeeper客户端连接服务端，-server指定ip:port ./zkCli.sh -server localhost:2181 # 如果登录连接本机，可以省略不写 ./zkCli.sh
  1. quit退出客户端，断开连接。
  2. ls 结点路径查看该结点下的子结点。

  1. create 结点目录 结点数据根据路径创建该结点，并指定该结点数据，如果未指定该结点数据，则为null；若相同结点路径已经存在，则不能创建相同结点路径。
  2. get 结点路径查看该结点的数据。
  3. set 结点路径 结点数据为该结点设置数据。
  4. delete 结点路径删除该结点，如果该结点下存在子结点，则无法直接删除多层结点。
  5. deleteall 多层结点路径删除带有子结点的结点。

3. 创建临时顺序结点

  1. create -e 结点路径 结点数据创建临时结点，只在当前会话窗口内有效，quit断开连接后无效。
  2. create -s 结点路径 结点数据创建顺序结点，该结点后面会自动加上一个数字编号，所有的结点都使用同一套数字编号。
  3. ls -s 结点路径或ls2 结点路径查询结点详细信息。

czxid：节点被创建的事务ID。

dataversion：数据版本号。

ctime：创建时间。

aclversion：权限版本号。

mzxid：最后一次被更新的事务ID。

ephemeralOwner：用于临时节点，代表临时节点的事务ID，如果为持久节点则为0。

mtime：修改时间。

pzxid：子节点列表最后一次被更新的事务ID。

dataLength：节点存储的数据的长度。

cversion：子节点的版本号。

numChildren：当前节点的子节点个数。

七、JavaAPI操作

1. Curator介绍

Curator是 Apache ZooKeeper 的Java客户端库。常见的ZooKeeper Java APl：

  • 原生Java APl
  • ZkClient
  • Curator

Curator项目的目标是简化ZooKeeper客户端的使用。

Curator最初是Netfix研发的，后来捐献了Apache基金会，目前是Apache的顶级项目。

官网：http://curator.apache.org/（zk版本3.5+，curator版本需要用4.0+）

  • 建立连接
  • 添加节点
  • 删除节点
  • 修改节点
  • 查询节点
  • Watch事件监听
  • 分布式锁实现

2. 建立连接

相关方法

解释说明

static CuratorFrameworkFactory.newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)

第一种建立客户端连接的方式，通过方法参数设置连接信息。

CuratorFrameworkFactory.builder().connectString(String connectString).sessionTimeoutMs(int sessionTimeoutMs).connectionTimeoutMs(int connectionTimeoutMs).retryPolicy(RetryPolicy retryPolicy).namespace(String path).build();

第二种建立客户端连接的方式，通过链式编程设置连接信息。

1）创建一个空项目zk-pro，里面创建一个maven项目curator-zk。

2）导入pom.xml和log4j.properties

<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">  <modelVersion>4.0.0</modelVersion>  <groupId>com.itheima</groupId>  <artifactId>curator-zk</artifactId>  <version>1.0-SNAPSHOT</version>  <dependencies>  <dependency>  <groupId>junit</groupId>  <artifactId>junit</artifactId>  <version>4.10</version>  <scope>test</scope>  </dependency>  <!-- curator -->  <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>  </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> </project>
log4j.rootLogger=off,stdout log4j.appender.stdout = org.apache.log4j.ConsoleAppender log4j.appender.stdout.Target = System.out log4j.appender.stdout.layout = org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern = [%d{yyyy-MM-dd HH/:mm/:ss}]%-5p %c(line/:%L) %x-%m%n

3）创建测试类CuratorTest，编写测试建立连接方法。

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.junit.Test; public class CuratorTest {  /**  * 建立客户端连接 Create a new client  */  @Test  public void testConnect() {  /*  第一种方式：  static CuratorFrameworkFactory.newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)  @param connectString 连接字符串，zk server的地址和端口 "192.168.8.100:2181,192.168.8.101:2181,..."  @param sessionTimeoutMs 会话超时时间，单位ms  @param connectionTimeoutMs 连接超时时间，单位ms  @param retryPolicy 重试策略  @return client  */  // 重试策略  RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10); // 3秒重试一次，最多重试10次  CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.8.100:2181", 60 * 1000, 15 * 1000, retryPolicy);  // 开启连接  client.start();  /*  第二种方式：链式编程  static CuratorFrameworkFactory.builder()  @param connectString 连接字符串，zk server的地址和端口 "192.168.8.100:2181,192.168.8.101:2181,..."  @param sessionTimeoutMs 会话超时时间，单位ms  @param connectionTimeoutMs 连接超时时间，单位ms  @param retryPolicy 重试策略  @return client  */  CuratorFramework client2 = CuratorFrameworkFactory.builder().connectString("192.168.8.100:2181")  .sessionTimeoutMs(60 * 1000)  .connectionTimeoutMs(15 * 1000)  .retryPolicy(new ExponentialBackoffRetry(3000, 10))  .namespace("itheima").build();  // 开启连接  client2.start();  } }

源码解读

Ctrl+左键点击newClient()静态方法跟进查看。

/** * Create a new client * * @param connectString list of servers to connect to * @param sessionTimeoutMs session timeout * @param connectionTimeoutMs connection timeout * @param retryPolicy retry policy to use * @return client */ public static CuratorFramework newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy) {  return builder().  connectString(connectString).  sessionTimeoutMs(sessionTimeoutMs).  connectionTimeoutMs(connectionTimeoutMs).  retryPolicy(retryPolicy).  build(); }

方法内部其实是调用了builder()方法做了链式编程，该方法参数有4个，分别是：

方法参数

解释说明

String connectString

连接字符串，zk server的地址和端口，可集群填写"192.168.8.100:2181,192.168.8.101:2181,..."

int sessionTimeoutMs

会话超时时间，单位ms

int connectionTimeoutMs

连接超时时间，单位ms

RetryPolicy retryPolicy

重试策略

对于重试策略RetryPolicy，继续跟进查看源码，发现这其实是一个接口。

/**  * Abstracts the policy to use when retrying connections  */ public interface RetryPolicy {  /**  * Called when an operation has failed for some reason. This method should return  * true to make another attempt.  *  *  * @param retryCount the number of times retried so far (0 the first time)  * @param elapsedTimeMs the elapsed time in ms since the operation was attempted  * @param sleeper use this to sleep - DO NOT call Thread.sleep  * @return true/false  */  public boolean allowRetry(int retryCount, long elapsedTimeMs, RetrySleeper sleeper); }

那么如何知道它的实现类呢？这里有个小技巧：点击其左侧的绿色接口小图标。

  • 对于接口，它会提示出该接口的实现类对象。
  • 对于接口内的抽象方法，它会提示出在哪个实现类中对该方法做了实现。

除此之外的小技巧，还有Ctrl+F12，可以查看当前类或接口的家族继承关系结构。

另外，对于第二种方式中的namespace("xxx")设置命名空间，这个其实是相当于给根结点设置一个根结点路径前缀/xxx，多个应用时设置名称空间起到隔离的作用，同时方便管理。

4）测试运行该测试类，发现绿了就说明测试成功，没问题！

3. 添加结点

方法参数

解释说明

create().forPath(String path)

基本创建结点。

create().forPath(String path, byte[] data)

创建结点并设置数据。

create().withMode(CreateMode mode).forPath(String path)

通过枚举类CreateMode设置结点的类型。如：持久结点，临时结点，顺序结点，可进行模式组合。

create().creatingParentsIfNeeded().forPath(String MultiLevelPath)

创建多级结点，如果父节点不存在，则创建父节点。

 
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.junit.After; import org.junit.Before; import org.junit.Test; public class CuratorTest {  private CuratorFramework client;  /**  * 建立客户端连接 Create a new client  */  @Before  public void testConnectClient() {  /*  第一种方式：  static CuratorFrameworkFactory.newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)  @param connectString 连接字符串，zk server的地址和端口 "192.168.8.100:2181,192.168.8.101:2181,..."  @param sessionTimeoutMs 会话超时时间，单位ms  @param connectionTimeoutMs 连接超时时间，单位ms  @param retryPolicy 重试策略  @return client  */  // 重试策略  /*RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10); // 3秒重试一次，最多重试10次  client = CuratorFrameworkFactory.newClient("192.168.8.100:2181", 60 * 1000, 15 * 1000, retryPolicy);  // 开启连接  client.start();*/  /*  第二种方式：链式编程  static CuratorFrameworkFactory.builder()  @param connectString 连接字符串，zk server的地址和端口 "192.168.8.100:2181,192.168.8.101:2181,..."  @param sessionTimeoutMs 会话超时时间，单位ms  @param connectionTimeoutMs 连接超时时间，单位ms  @param retryPolicy 重试策略  @return client  */  client = CuratorFrameworkFactory.builder().connectString("192.168.8.100:2181")  .sessionTimeoutMs(60 * 1000)  .connectionTimeoutMs(15 * 1000)  .retryPolicy(new ExponentialBackoffRetry(3000, 10))  .namespace("itheima").build();  // 开启连接  client.start();  }  /**  * 创建节点：create (持久 临时 顺序 可进行模式组合) 数据  * 1．基本创建 create().forPath("/xxx");  * 2．创建节点，带有数据 create().forPath("/xxx", dataOfBytesArray);  * 3．设器节点的类型 create().withMode(CreateMode.XXX).forPath("/xxx");  * 4、创建多级节点 create().creatingParentsIfNeeded().forPath("/xxx/xxx");  */  @Test  public void testCreateNode1() throws Exception {  // 1．基本创建  String path = client.create().forPath("/app1");  System.out.println(path);  }  @Test  public void testCreateNode2() throws Exception {  // 2．创建节点，带有数据  // 如果创建结点，没有指定数据，则默认将当前客户端的ip作为数据存储  String path = client.create().forPath("/app2", "haha".getBytes());  System.out.println(path);  }  @Test  public void testCreateNode3() throws Exception {  // 3．设器节点的类型  // 默认类型：持久化。如果该会话结束，则临时结点被会删除。注意：zkClient和zkJavaAPI属于两次不同的会话。  String path = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3"); // CreateMode是个枚举类  System.out.println(path);  }  @Test  public void testCreateNode4() throws Exception {  // 4、创建多级节点 /app4/p1  // creatingParentsIfNeeded()：如果父节点不存在，则创建父节点。  String path = client.create().creatingParentsIfNeeded().forPath("/app4/p1");  System.out.println(path);  }  @After  public void close() {  if (client != null) {  client.close();  }  } }

@Before和@After分别会在@Test测试方法的前后执行。

4. 查询结点

方法参数

解释说明

getData().forPath(String path)

查询数据：get。

getChildren().forPath(String path)

查询子结点：ls。

getData().storingStatIn(Stat stat要封装的状态对象).forPath(String path)

查询结点状态信息：ls -s。

 
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; import org.junit.After; import org.junit.Before; import org.junit.Test; import java.nio.charset.StandardCharsets; import java.util.List; public class CuratorTest { //==================================【Connect Client】==================================================  private CuratorFramework client;  /**  * 建立客户端连接 Create a new client  */  @Before  public void testConnectClient() {  /*  第一种方式：  static CuratorFrameworkFactory.newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)  @param connectString 连接字符串，zk server的地址和端口 "192.168.8.100:2181,192.168.8.101:2181,..."  @param sessionTimeoutMs 会话超时时间，单位ms  @param connectionTimeoutMs 连接超时时间，单位ms  @param retryPolicy 重试策略  @return client  */  // 重试策略  /*RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10); // 3秒重试一次，最多重试10次  client = CuratorFrameworkFactory.newClient("192.168.8.100:2181", 60 * 1000, 15 * 1000, retryPolicy);  // 开启连接  client.start();*/  /*  第二种方式：链式编程  static CuratorFrameworkFactory.builder()  @param connectString 连接字符串，zk server的地址和端口 "192.168.8.100:2181,192.168.8.101:2181,..."  @param sessionTimeoutMs 会话超时时间，单位ms  @param connectionTimeoutMs 连接超时时间，单位ms  @param retryPolicy 重试策略  @return client  */  client = CuratorFrameworkFactory.builder().connectString("192.168.8.100:2181")  .sessionTimeoutMs(60 * 1000)  .connectionTimeoutMs(15 * 1000)  .retryPolicy(new ExponentialBackoffRetry(3000, 10))  .namespace("itheima").build();  // 开启连接  client.start();  } //==================================【Create Node】==================================================  /**  * 创建节点：create (持久 临时 顺序 4种组合模式) 数据  * 1．基本创建 create().forPath("/xxx");  * 2．创建节点，带有数据 create().forPath("/xxx", dataOfBytesArray);  * 3．设器节点的类型 create().withMode(CreateMode.XXX).forPath("/xxx");  * 4、创建多级节点 create().creatingParentsIfNeeded().forPath("/xxx/xxx");  */  @Test  public void testCreateNode1() throws Exception {  // 1．基本创建  String path = client.create().forPath("/app1");  System.out.println(path);  }  @Test  public void testCreateNode2() throws Exception {  // 2．创建节点，带有数据  // 如果创建结点，没有指定数据，则默认将当前客户端的ip作为数据存储  String path = client.create().forPath("/app2", "haha".getBytes());  System.out.println(path);  }  @Test  public void testCreateNode3() throws Exception {  // 3．设器节点的类型  // 默认类型：持久化。如果该会话结束，则临时结点被会删除。注意：zkClient和zkJavaAPI属于两次不同的会话。  String path = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3"); // CreateMode是个枚举类  System.out.println(path);  }  @Test  public void testCreateNode4() throws Exception {  // 4、创建多级节点 /app4/p1  // creatingParentsIfNeeded()：如果父节点不存在，则创建父节点。  String path = client.create().creatingParentsIfNeeded().forPath("/app4/p1");  System.out.println(path);  } //==================================【Get Node】==================================================  /**  * 查询结点：  * 1. 查询数据：get getData().forPath("/xxx")  * 2. 查询子结点：ls getChildren().forPath("/")  * 3. 查询结点状态信息：ls -s getData().storingStatIn(要封装的状态对象).forPath("/xxx");  */  @Test  public void testGetNode1() throws Exception {  // 1. 查询数据：get  byte[] data = client.getData().forPath("/app1");  System.out.println(new String(data)); // 默认data存储的是当前ip地址 192.168.1.1  }  @Test  public void testGetNode2() throws Exception {  // 2. 查询子结点：ls  //List<String> childrenNodes = client.getChildren().forPath("/app4");  List<String> childrenNodes = client.getChildren().forPath("/"); // /itheima（这里查的是命名空间下的子结点，不是根目录下的子结点）  System.out.println(childrenNodes); // [app2, app1, app4]  }  @Test  public void testGetNode3() throws Exception {  // 3. 查询结点状态信息：ls -s  Stat status = new Stat();  System.out.println(status); // 0,0,0,0,0,0,0,0,0,0,0  byte[] data = client.getData().storingStatIn(status).forPath("/app1");  System.out.println(new String(data)); // 192.168.1.1  System.out.println(status); // 289,289,1677463941055,1677463941055,0,0,0,0,11,0,289  } //==================================【Close Client】==================================================  @After  public void close() {  if (client != null) {  client.close();  }  } }

5. 设置结点

方法参数

解释说明

setData().forPath("/xxx", "data".getBytes())

修改结点数据：set。

setData().withVersion(int version).forPath(String path, byte[] data)

根据版本修改结点数据：多个zk客户端并发操作时，只能等一个版本修改完之后更新版本，不同版本才能继续修改数据，相当于锁机制。

 
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; import org.junit.After; import org.junit.Before; import org.junit.Test; import java.nio.charset.StandardCharsets; import java.util.List; public class CuratorTest { //==================================【Connect Client】==================================================  private CuratorFramework client;  /**  * 建立客户端连接 Create a new client  */  @Before  public void testConnectClient() {  /*  第一种方式：  static CuratorFrameworkFactory.newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)  @param connectString 连接字符串，zk server的地址和端口 "192.168.8.100:2181,192.168.8.101:2181,..."  @param sessionTimeoutMs 会话超时时间，单位ms  @param connectionTimeoutMs 连接超时时间，单位ms  @param retryPolicy 重试策略  @return client  */  // 重试策略  /*RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10); // 3秒重试一次，最多重试10次  client = CuratorFrameworkFactory.newClient("192.168.8.100:2181", 60 * 1000, 15 * 1000, retryPolicy);  // 开启连接  client.start();*/  /*  第二种方式：链式编程  static CuratorFrameworkFactory.builder()  @param connectString 连接字符串，zk server的地址和端口 "192.168.8.100:2181,192.168.8.101:2181,..."  @param sessionTimeoutMs 会话超时时间，单位ms  @param connectionTimeoutMs 连接超时时间，单位ms  @param retryPolicy 重试策略  @return client  */  client = CuratorFrameworkFactory.builder().connectString("192.168.8.100:2181")  .sessionTimeoutMs(60 * 1000)  .connectionTimeoutMs(15 * 1000)  .retryPolicy(new ExponentialBackoffRetry(3000, 10))  .namespace("itheima").build();  // 开启连接  client.start();  } //==================================【Create Node】==================================================  /**  * 创建节点：create (持久 临时 顺序 可模式组合) 数据  * 1．基本创建 create().forPath("/xxx");  * 2．创建节点，带有数据 create().forPath("/xxx", dataOfBytesArray);  * 3．设置节点的类型 create().withMode(CreateMode.XXX).forPath("/xxx");  * 4、创建多级节点 create().creatingParentsIfNeeded().forPath("/xxx/xxx");  */  @Test  public void testCreateNode1() throws Exception {  // 1．基本创建  String path = client.create().forPath("/app1");  System.out.println(path);  }  @Test  public void testCreateNode2() throws Exception {  // 2．创建节点，带有数据  // 如果创建结点，没有指定数据，则默认将当前客户端的ip作为数据存储  String path = client.create().forPath("/app2", "haha".getBytes());  System.out.println(path);  }  @Test  public void testCreateNode3() throws Exception {  // 3．设器节点的类型  // 默认类型：持久化。如果该会话结束，则临时结点被会删除。注意：zkClient和zkJavaAPI属于两次不同的会话。  String path = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3"); // CreateMode是个枚举类  System.out.println(path);  }  @Test  public void testCreateNode4() throws Exception {  // 4、创建多级节点 /app4/p1  // creatingParentsIfNeeded()：如果父节点不存在，则创建父节点。  String path = client.create().creatingParentsIfNeeded().forPath("/app4/p1");  System.out.println(path);  } //==================================【Get Node】==================================================  /**  * 查询结点：  * 1. 查询数据：get getData().forPath("/xxx")  * 2. 查询子结点：ls getChildren().forPath("/")  * 3. 查询结点状态信息：ls -s getData().storingStatIn(要封装的状态对象).forPath("/xxx");  */  @Test  public void testGetNode1() throws Exception {  // 1. 查询数据：get  byte[] data = client.getData().forPath("/app1");  System.out.println(new String(data)); // 默认data存储的是当前ip地址 192.168.1.1  }  @Test  public void testGetNode2() throws Exception {  // 2. 查询子结点：ls  //List<String> childrenNodes = client.getChildren().forPath("/app4");  List<String> childrenNodes = client.getChildren().forPath("/"); // /itheima（这里查的是命名空间下的子结点，不是根目录下的子结点）  System.out.println(childrenNodes); // [app2, app1, app4]  }  @Test  public void testGetNode3() throws Exception {  // 3. 查询结点状态信息：ls -s  Stat status = new Stat();  System.out.println(status); // 0,0,0,0,0,0,0,0,0,0,0  byte[] data = client.getData().storingStatIn(status).forPath("/app1");  System.out.println(new String(data)); // 192.168.1.1  System.out.println(status); // 289,289,1677463941055,1677463941055,0,0,0,0,11,0,289  } //==================================【Set Node】==================================================  /**  * 修改数据  * 1. 修改数据 setData().forPath("/xxx", "数据".getBytes(StandardCharsets.UTF_8));  * 2. 根据版本修改 setData().withVersion(version).forPath("/app1", "数据".getBytes(StandardCharsets.UTF_8));  */  @Test  public void testSetNode() throws Exception {  client.setData().forPath("/app1", "itcast".getBytes());  }  @Test  public void testSetNodeForVersion() throws Exception {  Stat status = new Stat();  byte[] dataBefore = client.getData().storingStatIn(status).forPath("/app1");  System.out.println("修改前的数据：" + new String(dataBefore)); // itcast  int version = status.getVersion();  System.out.println("修改前的版本：" + version); // 1  status = client.setData().withVersion(version).forPath("/app1", "数据".getBytes(StandardCharsets.UTF_8));  System.out.println("修改后的版本：" + status.getVersion()); // 2  byte[] dataAfter = client.getData().forPath("/app1");  System.out.println("修改后的数据：" + new String(dataAfter)); // 数据  } //==================================【Close Client】==================================================  @After  public void close() {  if (client != null) {  client.close();  }  } }

6. 删除结点

方法参数

解释说明

delete().forPath(String path)

删除单个节点。

delete().deletingChildrenIfNeeded().forPath(String path)

删除带有子节点的节点。

delete().guaranteed().forPath(String path)

必须保证成功删除结点，为了防止网路抖动导致操作失败。本质是多次重试删除。

inBackground(BackgroundCallback callback)

删除后的回调操作，参数是一个回调接口，可传Lambda表达式。

 
import org.apache.curator.RetryPolicy; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.framework.api.BackgroundCallback; import org.apache.curator.framework.api.CuratorEvent; import org.apache.curator.retry.ExponentialBackoffRetry; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.data.Stat; import org.junit.After; import org.junit.Before; import org.junit.Test; import java.nio.charset.StandardCharsets; import java.util.List; public class CuratorTest { //==================================【Connect Client】==================================================  private CuratorFramework client;  /**  * 建立客户端连接 Create a new client  */  @Before  public void testConnectClient() {  /*  第一种方式：  static CuratorFrameworkFactory.newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)  @param connectString 连接字符串，zk server的地址和端口 "192.168.8.100:2181,192.168.8.101:2181,..."  @param sessionTimeoutMs 会话超时时间，单位ms  @param connectionTimeoutMs 连接超时时间，单位ms  @param retryPolicy 重试策略  @return client  */  // 重试策略  /*RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10); // 3秒重试一次，最多重试10次  client = CuratorFrameworkFactory.newClient("192.168.8.100:2181", 60 * 1000, 15 * 1000, retryPolicy);  // 开启连接  client.start();*/  /*  第二种方式：链式编程  static CuratorFrameworkFactory.builder()  @param connectString 连接字符串，zk server的地址和端口 "192.168.8.100:2181,192.168.8.101:2181,..."  @param sessionTimeoutMs 会话超时时间，单位ms  @param connectionTimeoutMs 连接超时时间，单位ms  @param retryPolicy 重试策略  @return client  */  client = CuratorFrameworkFactory.builder().connectString("192.168.8.100:2181")  .sessionTimeoutMs(60 * 1000)  .connectionTimeoutMs(15 * 1000)  .retryPolicy(new ExponentialBackoffRetry(3000, 10))  .namespace("itheima").build();  // 开启连接  client.start();  } //==================================【Create Node】==================================================  /**  * 创建节点：create (持久 临时 顺序 可模式组合) 数据  * 1．基本创建 create().forPath("/xxx");  * 2．创建节点，带有数据 create().forPath("/xxx", dataOfBytesArray);  * 3．设置节点的类型 create().withMode(CreateMode.XXX).forPath("/xxx");  * 4、创建多级节点 create().creatingParentsIfNeeded().forPath("/xxx/xxx");  */  @Test  public void testCreateNode1() throws Exception {  // 1．基本创建  String path = client.create().forPath("/app1");  System.out.println(path);  }  @Test  public void testCreateNode2() throws Exception {  // 2．创建节点，带有数据  // 如果创建结点，没有指定数据，则默认将当前客户端的ip作为数据存储  String path = client.create().forPath("/app2", "haha".getBytes());  System.out.println(path);  }  @Test  public void testCreateNode3() throws Exception {  // 3．设器节点的类型  // 默认类型：持久化。如果该会话结束，则临时结点被会删除。注意：zkClient和zkJavaAPI属于两次不同的会话。  String path = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3"); // CreateMode是个枚举类  System.out.println(path);  }  @Test  public void testCreateNode4() throws Exception {  // 4、创建多级节点 /app4/p1  // creatingParentsIfNeeded()：如果父节点不存在，则创建父节点。  String path = client.create().creatingParentsIfNeeded().forPath("/app4/p1");  System.out.println(path);  } //==================================【Get Node】==================================================  /**  * 查询结点：  * 1. 查询数据：get getData().forPath("/xxx")  * 2. 查询子结点：ls getChildren().forPath("/")  * 3. 查询结点状态信息：ls -s getData().storingStatIn(要封装的状态对象).forPath("/xxx");  */  @Test  public void testGetNode1() throws Exception {  // 1. 查询数据：get  byte[] data = client.getData().forPath("/app1");  System.out.println(new String(data)); // 默认data存储的是当前ip地址 192.168.1.1  }  @Test  public void testGetNode2() throws Exception {  // 2. 查询子结点：ls  //List<String> childrenNodes = client.getChildren().forPath("/app4");  List<String> childrenNodes = client.getChildren().forPath("/"); // /itheima（这里查的是命名空间下的子结点，不是根目录下的子结点）  System.out.println(childrenNodes); // [app2, app1, app4]  }  @Test  public void testGetNode3() throws Exception {  // 3. 查询结点状态信息：ls -s  Stat status = new Stat();  System.out.println(status); // 0,0,0,0,0,0,0,0,0,0,0  byte[] data = client.getData().storingStatIn(status).forPath("/app1");  System.out.println(new String(data)); // 192.168.1.1  System.out.println(status); // 289,289,1677463941055,1677463941055,0,0,0,0,11,0,289  } //==================================【Set Node】==================================================  /**  * 修改数据  * 1. 修改数据 setData().forPath("/xxx", "数据".getBytes(StandardCharsets.UTF_8));  * 2. 根据版本修改 setData().withVersion(version).forPath("/app1", "数据".getBytes(StandardCharsets.UTF_8));  */  @Test  public void testSetNode() throws Exception {  client.setData().forPath("/app1", "itcast".getBytes());  }  @Test  public void testSetNodeForVersion() throws Exception {  Stat status = new Stat();  byte[] dataBefore = client.getData().storingStatIn(status).forPath("/app1");  System.out.println("修改前的数据：" + new String(dataBefore)); // itcast  int version = status.getVersion();  System.out.println("修改前的版本：" + version); // 1  status = client.setData().withVersion(version).forPath("/app1", "数据".getBytes(StandardCharsets.UTF_8));  System.out.println("修改后的版本：" + status.getVersion()); // 2  byte[] dataAfter = client.getData().forPath("/app1");  System.out.println("修改后的数据：" + new String(dataAfter)); // 数据  } //==================================【Delete Node】==================================================  /**  * 删除节点：delete deLeteaLL  * 1. 删除单个节点 delete().forPath("/xxx");  * 2．删除带有子节点的节点 delete().deletingChildrenIfNeeded().forPath("/xxx");  * 3．保证成功删除结点 delete().guaranteed().forPath("/xxx");  * 4．删除后的回调操作 inBackground(BackgroundCallback callback 回调接口，可传Lambda表达式)  */  @Test  public void testDeleteNode1() throws Exception {  // 1. 删除单个节点  client.delete().forPath("/app1");  }  @Test  public void testDeleteNode2() throws Exception {  // 2．删除带有子节点的节点  client.delete().deletingChildrenIfNeeded().forPath("/app4");  }  @Test  public void testDeleteNode3() throws Exception {  // 3．必须保证成功删除结点，为了防止网路抖动导致操作失败。本质是多次重试删除  client.delete().guaranteed().forPath("/app2");  }  @Test  public void testDeleteNode4() throws Exception {  //client.create().forPath("/app1", "test".getBytes());  // 4．删除后的回调操作  client.delete().guaranteed().inBackground(new BackgroundCallback() {  @Override  public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {  System.out.println(client.checkExists());  System.out.println(event.getResultCode() == 0 ? "结点删除成功！" : "结点删除失败！");  System.out.println("client = " + client);  System.out.println("event = " + event);  }  }).forPath("/app1");  } //==================================【Close Client】==================================================  @After  public void close() {  if (client != null) {  client.close();  }  } }

7. Watch事件监听

ZooKeeper 允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，ZooKeeper服务端会将事件通知到感兴趣的客户端上去，该机制是ZooKeeper实现分布式协调服务的重要特性。

ZooKeeper中引入了Watcher机制来实现了发布/订阅功能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。

ZooKeeper原生支持通过注册Watcher来进行事件监听，但是其使用并不是特别方便需要开发人员自己反复注册Watcher，比较繁琐。

Curator引入了Cache来实现对ZooKeeper服务端事件的监听。ZooKeeper提供了三种Watcher：

  1. NodeCache：只是监听某一个特定的节点。
  2. PathChildrenCache：监控一个ZNode的子节点。
  3. TreeCache：可以监控整个树上的所有节点，类似于PathChildrenCache和NodeCache的组合。

（1）NodeCache

import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.framework.recipes.cache.*; import org.apache.curator.retry.ExponentialBackoffRetry; import org.junit.After; import org.junit.Before; import org.junit.Test; public class CuratorWatcherTest { //==================================【Connect Client】==================================================  private CuratorFramework client;  /**  * 建立客户端连接 Create a new client  */  @Before  public void testConnectClient() {  client = CuratorFrameworkFactory.builder().connectString("192.168.8.100:2181")  .sessionTimeoutMs(60 * 1000)  .connectionTimeoutMs(15 * 1000)  .retryPolicy(new ExponentialBackoffRetry(3000, 10))  .namespace("itheima").build();  // 开启连接  client.start();  } //==================================【NodeCache】==================================================  /**  * NodeCache：给指定一个结点注册监听器  */  @Test  public void testNodeCache() throws Exception {  // 1.创建NodeCache对象  final NodeCache nodeCache = new NodeCache(client, "/app1");  // 2.注册监听  nodeCache.getListenable().addListener(new NodeCacheListener() {  @Override  public void nodeChanged() throws Exception {  System.out.println("监听到结点发生变化了！");  // 获取修改结点后的数据  byte[] data = nodeCache.getCurrentData().getData();  System.out.println(new String(data));  }  });  // 3.开启监听，如果设置为true，则开启监听时预先加载缓存数据  nodeCache.start(true);  // 单元测试里模拟开了个线程持续监听，不退出  while (true);  } //==================================【Close Client】==================================================  @After  public void close() {  if (client != null) {  client.close();  }  } }

启动测试方法testNodeCache()，接下来在zkClient对/itheima/app1进行create、set、delete操作。

现在控制台发现对于增删改操作，该结点的监听器都能被监听到，并且可以监听结点数据的变化情况。

（2）PathChildrenCache

import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.framework.recipes.cache.*; import org.apache.curator.retry.ExponentialBackoffRetry; import org.junit.After; import org.junit.Before; import org.junit.Test; public class CuratorWatcherTest { //==================================【Connect Client】==================================================  private CuratorFramework client;  /**  * 建立客户端连接 Create a new client  */  @Before  public void testConnectClient() {  client = CuratorFrameworkFactory.builder().connectString("192.168.8.100:2181")  .sessionTimeoutMs(60 * 1000)  .connectionTimeoutMs(15 * 1000)  .retryPolicy(new ExponentialBackoffRetry(3000, 10))  .namespace("itheima").build();  // 开启连接  client.start();  } //==================================【NodeCache】==================================================  /**  * NodeCache：给指定一个结点注册监听器  */  @Test  public void testNodeCache() throws Exception {  // 1.创建NodeCache对象  final NodeCache nodeCache = new NodeCache(client, "/app1");  // 2.注册监听  nodeCache.getListenable().addListener(new NodeCacheListener() {  @Override  public void nodeChanged() throws Exception {  System.out.println("监听到结点发生变化了！");  // 获取修改结点后的数据  byte[] data = nodeCache.getCurrentData().getData();  System.out.println(new String(data));  }  });  // 3.开启监听，如果设置为true，则开启监听时预先加载缓存数据  nodeCache.start(true);  // 单元测试里模拟开了个线程持续监听，不退出  while (true);  } //==================================【PathChildrenCache】==================================================  /**  * PathChildrenCache：监听某个结点的所有子结点们  */  @Test  public void testPathChildrenCache() throws Exception {  // 1.创建监听对象  final PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/app2", true);  // 2.绑定监听器  pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {  @Override  public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {  System.out.println("监听到子结点变化了！");  System.out.println(event);  // 监听子结点的数据变更，并且拿到变更后的数据  // 获取操作类型  PathChildrenCacheEvent.Type type = event.getType();  // 判断类型是否是update  if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {  System.out.println("数据发生修改！");  byte[] data = event.getData().getData(); // 第一次getData获取的是ChildData对象，第二次getData才是存的data数据  System.out.println("变更后的数据：" + new String(data));  }  }  });  // 3.开启监听  pathChildrenCache.start();  // 单元测试里模拟开了个线程持续监听，不退出  while (true);  } //==================================【Close Client】==================================================  @After  public void close() {  if (client != null) {  client.close();  }  } }

启动测试方法testPathChildrenCache()，接下来在zkClient对/itheima/app2/p1进行create、set、delete操作。

从event对象中获取相关数据。

（3）TreeCache

import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.framework.recipes.cache.*; import org.apache.curator.retry.ExponentialBackoffRetry; import org.junit.After; import org.junit.Before; import org.junit.Test; public class CuratorWatcherTest { //==================================【Connect Client】==================================================  private CuratorFramework client;  /**  * 建立客户端连接 Create a new client  */  @Before  public void testConnectClient() {  client = CuratorFrameworkFactory.builder().connectString("192.168.8.100:2181")  .sessionTimeoutMs(60 * 1000)  .connectionTimeoutMs(15 * 1000)  .retryPolicy(new ExponentialBackoffRetry(3000, 10))  .namespace("itheima").build();  // 开启连接  client.start();  } //==================================【NodeCache】==================================================  /**  * NodeCache：给指定一个结点注册监听器  * 这个NodeCache在5.x版本弃置了，用CuratorCache代替了  */  @Test  public void testNodeCache() throws Exception {  // 1.创建NodeCache对象  final NodeCache nodeCache = new NodeCache(client, "/app1");  // 2.注册监听  nodeCache.getListenable().addListener(new NodeCacheListener() {  @Override  public void nodeChanged() throws Exception {  System.out.println("监听到结点发生变化了！");  // 获取修改结点后的数据  byte[] data = nodeCache.getCurrentData().getData();  System.out.println(new String(data));  }  });  // 3.开启监听，如果设置为true，则开启监听时预先加载缓存数据  nodeCache.start(true);  // 单元测试里模拟开了个线程持续监听，不退出  while (true);  } //==================================【PathChildrenCache】==================================================  /**  * PathChildrenCache：监听某个结点的所有子结点们  */  @Test  public void testPathChildrenCache() throws Exception {  // 1.创建监听对象  final PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/app2", true);  // 2.绑定监听器  pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {  @Override  public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {  System.out.println("监听到子结点变化了！");  System.out.println(event);  // 监听子结点的数据变更，并且拿到变更后的数据  // 获取操作类型  PathChildrenCacheEvent.Type type = event.getType();  // 判断类型是否是update  if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {  System.out.println("数据发生修改！");  byte[] data = event.getData().getData(); // 第一次getData获取的是ChildData对象，第二次getData才是存的data数据  System.out.println("变更后的数据：" + new String(data));  }  }  });  // 3.开启监听  pathChildrenCache.start();  // 单元测试里模拟开了个线程持续监听，不退出  while (true);  } //==================================【TreeCache】==================================================  /**  * TreeCache：监听某个结点自己和所有子结点们  */  @Test  public void testTreeCache() throws Exception {  // 1.创建监听器  final TreeCache treeCache = new TreeCache(client, "/app2");  // 2.注册监听  treeCache.getListenable().addListener(new TreeCacheListener() {  @Override  public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {  System.out.println("监听到子结点变化了！");  System.out.println(event);  TreeCacheEvent.Type type = event.getType();  if (type.equals(TreeCacheEvent.Type.NODE_UPDATED)) {  String path = event.getData().getPath();  byte[] data = event.getData().getData();  System.out.println(path + "结点的数据变更为：" + new String(data));  }  }  });  // 3.开启监听  treeCache.start();  // 单元测试里模拟开了个线程持续监听，不退出  while (true);  } //==================================【Close Client】==================================================  @After  public void close() {  if (client != null) {  client.close();  }  } }

测试监听的结点自己和子结点即可，和上面效果同理。

8. 分布式锁

（1）概念

  • 在我们进行单机应用开发，涉及并发同步的时候，我们往往采用synchronized或者Lock的方式来解决多线程间的代码同步问题，这时多线程的运行都是在同一个JVM之下，没有任何问题。

  • 但当我们的应用是分布式集群工作的情况下，属于多JVM下的工作环境，跨JVM之间已经无法通过多线程的锁解决同步问题。
  • 那么就需要一种更加高级的锁机制，来处理这种跨机器的进程之间的数据同步问题——这就是分布式锁。

  • 市面上常见的分布式锁实现的解决方案，如基于缓存实现分布式锁、Zookeeper实现分布式锁、数据库层面实现分布式锁。Redis靠setnx唯一节点，数据库靠版本，zk靠序号。

（2）Zookeeper分布式锁原理

核心思想：当客户端要获取锁，则创建节点，使用完锁，则删除该节点。

  1. 客户端获取锁时，在lock节点下创建临时顺序节点。
  2. 每个客户端来了都要获取lock下面的所有子节点，客户端获取到所有的子节点之后，如果发现自己创建的子节点序号最小，那么就认为该客户端获取到了锁。使用完锁后，将该节点删除。
  3. 如果发现自己创建的节点并非lock节点下所有子节点中最小的，说明自己还没有获取到锁，此时客户端需要找到比自己小的那个节点，同时对其注册事件监听器，监听删除事件。
  4. 如果发现比自己小的那个节点被删除，则客户端的Watcher会收到相应通知，此时再次判断自己创建的节点是否是lock子节点中序号最小的，如果是则获取到了锁，如果不是则重复以上步骤继续获取到比自己小的一个节点并注册监听。

（3）分布式锁案例：模拟12306售票

先来了解下Curator实现分布式锁的API，在Curator中有五种锁方案：

  1. lnterProcessSemaphoreMutex：分布式排它锁（非可重入锁）
  2. InterProcessMutex：分布式可重入排它锁
  3. InterProcessReadWriteLock：分布式读写锁
  4. InterProcessMultiLock：将多个锁作为单个实体管理的容器
  5. lnterProcessSemaphoreV2：共享信号量

1）创建Ticket12306卖票线程任务类。

public class Ticket12306 implements Runnable {  // 数据库的票数（共享资源）  private int ticket = 10;  @Override  public void run() {  while (true) {  if (ticket > 0) {  try {  Thread.sleep(1000);  } catch (InterruptedException e) {  e.printStackTrace();  }  System.out.println(Thread.currentThread().getName() + "@" + ticket);  ticket--;  }  }  } }

2）创建线程测试类DistributedLockTest。

public class DistributedLockTest {  public static void main(String[] args) {  // 创建线程任务对象  Ticket12306 ticket12306 = new Ticket12306();  // 创建卖票客户端，模拟卖票，启动线程  new Thread(ticket12306, "携程").start();  new Thread(ticket12306, "去哪儿").start();  } }

测试运行，出现重复票和超卖情况。

由于实际情况下Ticket12306卖票服务会集群部署很多份，不能通过同步代码块或JVM多线程提供的Lock锁方案，因此我们需要用第三方锁资源，即ZK提供的分布式锁解决该线程同步问题。

3）下面我们对卖票程序进行改造，加入分布式锁。

import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.framework.recipes.locks.InterProcessMutex; import org.apache.curator.retry.ExponentialBackoffRetry; import java.util.concurrent.TimeUnit; public class Ticket12306 implements Runnable {  // 数据库的票数（共享资源）  private int ticket = 10;  // 分布式锁对象  private final InterProcessMutex lock;  public Ticket12306() {  CuratorFramework client = CuratorFrameworkFactory.builder().connectString("192.168.8.100:2181")  .sessionTimeoutMs(60 * 1000)  .connectionTimeoutMs(15 * 1000)  .retryPolicy(new ExponentialBackoffRetry(3000, 10))  .build();  // 开启连接  client.start();  // 初始化分布式锁，在/lock下创建临时顺序结点  lock = new InterProcessMutex(client, "/lock");  }  @Override  public void run() {  while (true) {  try {  // 获取锁，3秒获取一次  lock.acquire(3, TimeUnit.SECONDS);  if (ticket > 0) {  Thread.sleep(100);  System.out.println(Thread.currentThread().getName() + "@" + ticket);  ticket--;  }  } catch (Exception e) {  e.printStackTrace();  }finally {  // 释放锁  try {  lock.release();  } catch (Exception e) {  e.printStackTrace();  }  }  }  } }

测试售票正常。

运行后发现自动创建了/lock结点，而/lock下创建有临时顺序结点，此时就会根据ZK分布式锁原理，结点编号较小的先获取锁。

八、集群搭建

1. 集群介绍

Leader选举：

  • Serverid：服务器ID
  • 比如有三台服务器，编号分别是1，2，3。编号越大在选择算法中的权重越大。
  • Zxid：数据lD
  • 服务器中存放的最大数据ID值越大说明数据越新，在选举算法中数据越新权重越大。
  • 在Leader选举的过程中，如果某台ZooKeeper获得了超过半数的选票，则此ZooKeeper就可以成为Leader了。

选举分析：在上图5个ZK顺序编号选举中，1号和2号会为3号投票，3票大于半数选票，ZK3号会作为Leader出现。如果是3个ZK顺序编号选举，则第二个ZK作为领导者。

2. 搭建要求

真实的集群是需要部署在不同的服务器上的，但是在我们测试时同时启动很多个虚拟机内存会吃不消，所以我们通常会搭建伪集群，也就是把所有的服务都搭建在一台虚拟机上，用端口进行区分。

我们这里要求搭建一个三个节点的Zookeeper集群（伪集群）。

3. 准备工作

部署一台虚拟机作为我们搭建集群的测试服务器。

  1. 安装JDK 【此步骤省略】。
  2. Zookeeper压缩包上传到服务器[root@localhost ~]#

  1. 将Zookeeper解压 ，建立/usr/local/zookeeper-cluster目录，将解压后的Zookeeper复制到以下三个目录
  • /usr/local/zookeeper-cluster/zookeeper-1
  • /usr/local/zookeeper-cluster/zookeeper-2
  • /usr/local/zookeeper-cluster/zookeeper-3
mkdir /usr/local/zookeeper-cluster cp -r apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-1 cp -r apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-2 cp -r apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-3 cd /usr/local/zookeeper-cluster
  1. 创建data目录 ，并且将conf下zoo_sample.cfg文件改名为zoo.cfg
mkdir /usr/local/zookeeper-cluster/zookeeper-1/data mkdir /usr/local/zookeeper-cluster/zookeeper-2/data mkdir /usr/local/zookeeper-cluster/zookeeper-3/data mv /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo_sample.cfg /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg mv /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo_sample.cfg /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg mv /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo_sample.cfg /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg
  1. 配置每一个Zookeeper的dataDir和clientPort分别为2181、2182、2183。真集群只需要改数据目录dataDir，伪集群则还需要分别改端口clientPort，区别三台不同ZK机器。

修改/usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg。

vim /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg clientPort=2181 dataDir=/usr/local/zookeeper-cluster/zookeeper-1/data

修改/usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg clientPort=2182 dataDir=/usr/local/zookeeper-cluster/zookeeper-2/data

修改/usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg clientPort=2183 dataDir=/usr/local/zookeeper-cluster/zookeeper-3/data

4. 配置集群

  1. 在每个zookeeper的 data 目录下创建一个 myid 文件，内容分别是1、2、3。这个文件就是记录每个服务器的ID。
echo 1 >/usr/local/zookeeper-cluster/zookeeper-1/data/myid echo 2 >/usr/local/zookeeper-cluster/zookeeper-2/data/myid echo 3 >/usr/local/zookeeper-cluster/zookeeper-3/data/myid
  1. 为了让每个zookeeper都知道对方的存在，需要在每一个zookeeper的 zoo.cfg 配置客户端访问端口（clientPort）和 集群服务器IP列表。

集群服务器IP列表如下

vim /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg vim /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg vim /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg # 伪集群 server.1=192.168.8.100:2881:3881 server.2=192.168.8.100:2882:3882 server.3=192.168.8.100:2883:3883 # 真集群搭建配置 #server.1=192.168.8.100:2881:3881 #server.2=192.168.8.101:2881:3881 #server.3=192.168.8.102:2881:3881

解释：server.服务器ID=服务器IP地址:服务器之间通信端口:服务器之间投票选举端口，之前的2181是客户端和服务端之间连接的默认端口，默认服务端之间相互通信的默认端口是2881，服务器之间投票选举的默认端口是3881。

5. 启动集群

启动集群就是分别启动每个实例。

# 分别启动每个ZK实例 /usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh start /usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh start /usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh start # 查看每个ZK的角色状态 /usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status /usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status /usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

注意：在此之前需要将其他单节点Zookeeper关掉。

集群搭建成功！此时2号的Mode为leader，1号和3号的Mode为follower。都不是之前单节点ZK的Mode：standalone了。

  • Mode为follower表示是跟随者（从）
  • Mode为leader表示是领导者（主）

6. 模拟集群异常

接下来我们来模拟一下集群异常测试，如果其中一个服务器挂掉会怎么样。

  1. 首先我们先测试如果是从服务器挂掉，会怎么样？

把3号服务器停掉，观察1号和2号，发现状态并没有变化。

/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh stop /usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status /usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

由此得出结论，3个节点的集群，从服务器挂掉，集群正常。

  1. 我们再把1号服务器（从服务器）也停掉，查看2号（主服务器）的状态，发现已经停止运行了。
/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh stop /usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

由此得出结论，3个节点的集群，2个从服务器都挂掉，主服务器也无法运行。因为可运行的机器没有超过集群总数量的半数。

  1. 我们再次把1号服务器启动起来，发现2号服务器又开始正常工作了。而且依然是领导者。
/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh start /usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

  1. 我们把3号服务器也启动起来，把2号服务器停掉,停掉后观察1号和3号的状态。
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh start /usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh stop /usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status /usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

发现新的 leader 产生了~

由此我们得出结论：当集群中的主服务器挂了，集群中的其他服务器会自动进行选举状态，然后产生新的leader。

  1. 我们再次测试，当我们把2号服务器重新启动起来启动后，会发生什么？2号服务器会再次成为新的领导吗？我们看结果。
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh start /usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status /usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

我们会发现，2号服务器启动后依然是跟随者（从服务器），3号服务器依然是领导者（主服务器），没有撼动3号服务器的领导地位。

由此我们得出结论：当领导者产生后，再次有新服务器加入集群，不会影响到现任领导者。

九、核心理论

在ZooKeeper集群服务中有三个角色：

  1. Leader领导者：
  • 处理事务请求（增删改）
  • 集群内部各服务器的调度者，处理完事务请求后为所有ZK同步数据。
  1. Follower跟随者：
  • 处理客户端非事务请求（查），转发事务请求给Leader服务器
  • 参与Leader选举投票
  1. Observer观察者：
  • 处理客户端非事务请求，转发事务请求给Leader服务器
文章标签：
微服务引擎
云解析DNS
Java
SpringCloudAlibaba
数据库
存储
Apache
关键词：
微服务引擎分布式
springcloud分布式
分布式集群
微服务引擎集群
springcloud alibaba
来 源：
开发者社区 > 数据库 > 文章 > 正文
蓝染-惣右介
目录
相关文章
阿里云云原生
|
28天前
|
人工智能 Java Nacos
基于 Spring AI Alibaba + Nacos 的分布式 Multi-Agent 构建指南
本文将针对 Spring AI Alibaba + Nacos 的分布式多智能体构建方案展开介绍，同时结合 Demo 说明快速开发方法与实际效果。
阿里云云原生
1312 52
啦啦啦191
|
3月前
|
监控 Java API
Spring Boot 3.2 结合 Spring Cloud 微服务架构实操指南 现代分布式应用系统构建实战教程
Spring Boot 3.2 + Spring Cloud 2023.0 微服务架构实践摘要 本文基于Spring Boot 3.2.5和Spring Cloud 2023.0.1最新稳定版本，演示现代微服务架构的构建过程。主要内容包括： 技术栈选择：采用Spring Cloud Netflix Eureka 4.1.0作为服务注册中心，Resilience4j 2.1.0替代Hystrix实现熔断机制，配合OpenFeign和Gateway等组件。 核心实操步骤： 搭建Eureka注册中心服务 构建商品
啦啦啦191
636 3
一枚后端工程狮
|
29天前
|
消息中间件 分布式计算 资源调度
《聊聊分布式》ZooKeeper与ZAB协议：分布式协调的核心引擎
ZooKeeper是一个开源的分布式协调服务，基于ZAB协议实现数据一致性，提供分布式锁、配置管理、领导者选举等核心功能，具有高可用、强一致和简单易用的特点，广泛应用于Kafka、Hadoop等大型分布式系统中。
一枚后端工程狮
125 3
一枚后端工程狮
|
1月前
|
负载均衡 Java API
《深入理解Spring》Spring Cloud 构建分布式系统的微服务全家桶
Spring Cloud为微服务架构提供一站式解决方案，涵盖服务注册、配置管理、负载均衡、熔断限流等核心功能，助力开发者构建高可用、易扩展的分布式系统，并持续向云原生演进。
一枚后端工程狮
429 4
阿里云云原生
|
6月前
|
人工智能 负载均衡 Java
Spring AI Alibaba 发布企业级 MCP 分布式部署方案
本文介绍了Spring AI Alibaba MCP的开发与应用，旨在解决企业级AI Agent在分布式环境下的部署和动态更新问题。通过集成Nacos，Spring AI Alibaba实现了流量负载均衡及节点变更动态感知等功能。开发者可方便地将企业内部业务系统发布为MCP服务或开发自己的AI Agent。文章详细描述了如何通过代理应用接入存量业务系统，以及全新MCP服务的开发流程，并提供了完整的配置示例和源码链接。未来，Spring AI Alibaba计划结合Nacos3的mcp-registry与mcp-router能力，进一步优化Agent开发体验。
阿里云云原生
2400 15
小陈运维
|
7月前
|
消息中间件 运维 Java
搭建Zookeeper、Kafka集群
本文详细介绍了Zookeeper和Kafka集群的搭建过程，涵盖系统环境配置、IP设置、主机名设定、防火墙与Selinux关闭、JDK安装等基础步骤。随后深入讲解了Zookeeper集群的安装与配置，包括数据目录创建、节点信息设置、SASL认证配置及服务启动管理。接着描述了Kafka集群的安装，涉及配置文件修改、安全认证设置、生产消费认证以及服务启停操作。最后通过创建Topic、发送与查看消息等测试验证集群功能。全网可搜《小陈运维》获取更多信息。
小陈运维
618 1
XiaoHHSuperme
|
10月前
|
Java 关系型数据库 数据库
微服务SpringCloud分布式事务之Seata
SpringCloud+SpringCloudAlibaba的Seata实现分布式事务，步骤超详细，附带视频教程
XiaoHHSuperme
764 1
阿里云开发者
|
12月前
|
存储 运维 NoSQL
分布式读写锁的奥义：上古世代 ZooKeeper 的进击
本文作者将介绍女娲对社区 ZooKeeper 在分布式读写锁实践细节上的思考，希望帮助大家理解分布式读写锁背后的原理。
阿里云开发者
311 11
赵渝强老师
|
3月前
|
存储 负载均衡 NoSQL
【赵渝强老师】Redis Cluster分布式集群
Redis Cluster是Redis的分布式存储解决方案，通过哈希槽（slot）实现数据分片，支持水平扩展，具备高可用性和负载均衡能力，适用于大规模数据场景。
赵渝强老师
297 2
别惹CC
|
3月前
|
存储 缓存 NoSQL
【📕分布式锁通关指南 12】源码剖析redisson如何利用Redis数据结构实现Semaphore和CountDownLatch
本文解析 Redisson 如何通过 Redis 实现分布式信号量（RSemaphore）与倒数闩（RCountDownLatch），利用 Lua 脚本与原子操作保障分布式环境下的同步控制，帮助开发者更好地理解其原理与应用。
别惹CC
237 6

热门文章

最新文章

  • 1
    基于 Spring AI Alibaba + Nacos 的分布式 Multi-Agent 构建指南
  • 2
    大模型分布式推理：张量并行与流水线并行技术
  • 3
    《聊聊分布式》BASE理论 分布式系统可用性与一致性的工程平衡艺术
  • 4
    《聊聊分布式》ZooKeeper与ZAB协议：分布式协调的核心引擎
  • 5
    Kali虚拟机间搭建iodine DNS隧道实验指南
  • 6
    别再手写过滤器！SpringCloud Gateway 内置30 个，少写 80% 重复代码
  • 7
    分布式 Multi Agent 安全高可用探索与实践
  • 8
    68_分布式训练技术：DDP与Horovod
  • 9
    《聊聊分布式》分布式系统基石：深入理解CAP理论及其工程实践
  • 10
    Apache Kafka 分布式流处理平台技术详解与实践指南
  • 1
    ​SpringCloud统一异常处理
    193
  • 2
    基于 Spring Cloud 的微服务架构分析
    594
  • 3
    尚硅谷SpringCloud教程 笔记
    444
  • 4
    Spring Cloud OpenFeign详解与实践
    491
  • 5
    SpringCloud调用组件Feign
    619
  • 6
    Spring Cloud五大组件
    3894
  • 7
    Spring Cloud Alibaba与Spring Cloud区别和联系？
    1407
  • 8
    SaaS云计算技术的智慧工地源码，基于Java+Spring Cloud框架开发
    196
  • 9
    SpringCloud Alibaba AI整合DeepSeek落地AI项目实战
    2915
  • 10
    springcloud/springboot集成NACOS 做注册和配置中心以及nacos源码分析
    3757

    • 相关课程

    查看更多
  • Spring Boot+Vue.js+FastDFS实现分布式图片服务器
  • 微服务+全栈在线教育实战项目演练（SpringCloud Alibaba+SpringBoot）
  • 基于Zookeeper、Dubbo构建互联网分布式基础架构
  • 大数据ZooKeeper快速入门
  • 分布式文件存储系统技术及实现
  • 分布式协调系统 Zookeeper 快速入门

    • 相关电子书

    查看更多
  • 高并发分布式缓存Redis6.0
  • 基于社区的分布式 风险感知模型
  • 如何利用Redisson分布式化传统Web项目

    • 推荐镜像

    查看更多
  • DNS
    • 下一篇