SpringCloud Alibaba系列全面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管理一些配置来实现服务治理。

2. 分布式锁

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

3. 集群管理

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

4. 生成分布式唯一ID

在过去的单库单表型系统中，通常可以使用数据库字段自带的auto_increment属性来自动为每条记录生成一个唯一的ID。
但是分库分表后，就无法在依靠数据库的auto_increment属性来唯一标识一条记录了。此时我们就可以用Zookeeper在分布式环境下生成全局唯一ID。
做法如下：每次要生成一个新id时，创建一个持久顺序节点，创建操作返回的节点序号，即为新id，然后把比自己节点小的删除即可。

---

三、设计目标

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

1. 高性能

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

2. 高可用

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

3. 严格顺序访问

对于来自客户端的每个更新请求，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

2. 配置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

3. 检测jdk安装java -version，如果反馈了Java信息，则成功。
4. 将下载的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/

5. 将tar包解压到/opt/zookeeper目录下。

tar -zxvf apache-zookeeper-3.5.6-bin.tar.gz -C /opt/zookeeper/

6. 为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

7. 2181是ZK监听客户端连接的默认端口号，记得防火墙把这个端口给开放出来，不然客户端连接不上。

# 添加2181端口到防火墙允许列表
firewall-cmd --zone=public --add-port=2181/tcp --permanent
# 立即生效
firewall-cmd --reload
# 查看Linux防火墙允许通过的端口号
firewall-cmd --zone=public --list-ports

8. 启动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

2. quit退出客户端，断开连接。
3. ls 结点路径查看该结点下的子结点。

4. create 结点目录 结点数据根据路径创建该结点，并指定该结点数据，如果未指定该结点数据，则为null；若相同结点路径已经存在，则不能创建相同结点路径。
5. get 结点路径查看该结点的数据。
6. set 结点路径 结点数据为该结点设置数据。
7. delete 结点路径删除该结点，如果该结点下存在子结点，则无法直接删除多层结点。
8. 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. 建立连接

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个，分别是：

对于重试策略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. 添加结点

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. 查询结点

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. 设置结点

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. 删除结点

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分布式锁原理

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

:::danger

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 ~]#

3. 将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

4. 创建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

5. 配置每一个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

2. 为了让每个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个节点的集群，从服务器挂掉，集群正常。

2. 我们再把1号服务器（从服务器）也停掉，查看2号（主服务器）的状态，发现已经停止运行了。

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

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

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

3. 我们再次把1号服务器启动起来，发现2号服务器又开始正常工作了。而且依然是领导者。

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh start

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

4. 我们把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。

5. 我们再次测试，当我们把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同步数据。

2. Follower跟随者：

• 处理客户端非事务请求（查），转发事务请求给Leader服务器
• 参与Leader选举投票

3. Observer观察者：

• 处理客户端非事务请求，转发事务请求给Leader服务器

---

这篇好文章是转载于：学新通技术网