Zookeeper ZkClient

1. 前言

无论是选择 Zookeeper 的单机模式还是选择 Zookeeper 的集群模式，实际开发中我们都会使用 Java 客户端向Zookeeper 服务端发送请求。本节我们就来学习如何使用 Zookeeper 的 Java 客户端之一 ZkClient。

2. ZkClient 客户端

ZkClient 是一个开源的客户端，在 Zookeeper 原生 API 接口的基础上进行了包装，更便于开发人员使用。内部实现了Session超时重连，Watcher反复注册等功能。想要使用 ZkClient，我们需要搭建 Java 开发环境，这里我们使用 IntelliJ IDEA 为开发工具，JDK 我们使用了长期维护的版本 JDK-11.0.8 。接下来我们开始搭建 ZkClient 运行的环境。

2.1 ZkClient 的依赖

首先我们新建 Spring Boot 项目，在 pom.xml 文件中加入 zkclient 的依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.2.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>cn.cdd</groupId>
    <artifactId>zkclient-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>zkclient-demo</name>
    <description>zkclient-demo project for Spring Boot</description>

    <properties>
        <java.version>11</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <!-- https://mvnrepository.com/artifact/com.101tec/zkclient -->
        <dependency>
            <groupId>com.101tec</groupId>
            <artifactId>zkclient</artifactId>
            <version>0.11</version>
            <!-- 排除冲突 -->
            <exclusions>
                <exclusion>
                    <artifactId>log4j</artifactId>
                    <groupId>log4j</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>slf4j-log4j12</artifactId>
                    <groupId>org.slf4j</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>slf4j-api</artifactId>
                    <groupId>org.slf4j</groupId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

加入了 ZkClient的依赖，我们就可以编写程序使用 zkclient 的 API 来向 Zookeeper 服务端发送请求了。

2.2 编写 ZkClientServer 类

我们在 Spring Boot 主函数的同级新建目录 service ，在 service 目录中新建类 ZkClientServer ：

package cn.cdd.zkclientdemo.service;

import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
// 声明 ZkClientServer 类为 Bean，交给 Spring 管理
@Component
public class ZkClientServer {
	// Zookeeper 集群的地址
    @Value(value = "${zookeeper.address}")
    private String address;
	// 获取 ZkClient
    public ZkClient getZkClient() {
        return new ZkClient(address);
    }
}

在 application.properties 配置文件中配置 Zookeeper 集群的地址：

zookeeper.address=192.168.0.77:2181,192.168.0.88:2181,192.168.0.99:2181

配置完成后，我们就可以去测试类中进行测试了。

2.3 测试

我们在项目中与 main 目录同级的 test 目录中找到测试类 ZkClientDemoApplicationTests ，在其中添加测试方法。

2.3.1 查询测试

package cn.cdd.zkclientdemo;

import cn.cdd.zkclientdemo.service.ZkClientServer;
import org.I0Itec.zkclient.ZkClient;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.List;

@SpringBootTest
class ZkClientDemoApplicationTests {
	// 依赖注入
    @Autowired
    private ZkClientServer zkClientServer;
    
    @Test
    void contextLoads() {
        // 获取 ZkClient 对象
        ZkClient zkClient = zkClientServer.getZkClient();
        // 获取子节点集合
        List<String> children = zkClient.getChildren("/");
        System.out.println(children);
        // 释放资源
        zkClient.close();
    }
}

执行测试方法，控制台输出：

[zookeeper, imooc]

我们可以看到控制台输出了在上一节中我们增加的节点 imooc，说明我们的查询成功执行了。接下来我们测试其它的 API。

2.3.2 创建持久节点

创建持久节点测试：

@Test
void contextLoads() {
    ZkClient zkClient = zkClientServer.getZkClient();
    // 在 imooc 节点下创建持久节点 wiki
    zkClient.createPersistent("/imooc/wiki");
    // 获取 imooc 节点的子节点集合
    List<String> children = zkClient.getChildren("/imooc");
    System.out.println(children);
    // 释放资源
    zkClient.close();
}

执行测试方法，控制台输出：

[wiki]

2.3.3 删除节点

删除节点测试：

@Test
void contextLoads() {
    ZkClient zkClient = zkClientServer.getZkClient();
    // 删除 imooc 的 子节点 wiki
    boolean delete = zkClient.delete("/imooc/wiki");
    System.out.println(delete);
    // 释放资源
    zkClient.close();
}

执行测试方法，控制台输出：

true

true 表示删除 wiki 节点成功。

2.3.4 读写数据

节点数据读写测试：

@Test
void contextLoads() {
    ZkClient zkClient = zkClientServer.getZkClient();
    // 给 imooc 节点写入数据 wiki
    zkClient.writeData("/imooc","wiki");
    // 读取 imooc 节点的数据
    Object data = zkClient.readData("/imooc");
    System.out.println(data);
    // 释放资源
    zkClient.close();
}

执行测试方法，控制台输出：

wiki

Tips： 当我们使用 API 操作节点时，节点参数必须是全路径。

测试完成后，我们来对 ZkClient 常用的 API 做一下介绍。

3. ZkClient API

• ZkClient 初始化 API

// serverstring：服务器地址的字符串
// 如：192.168.0.77:2181,192.168.0.88:2181,192.168.0.99:2181
public ZkClient(String serverstring);
// zkServers：服务器地址的字符串，connectionTimeout：连接超时时间，单位：ms
public ZkClient(String zkServers, int connectionTimeout);
// sessionTimeout： session 超时时间
public ZkClient(String zkServers, int sessionTimeout, int connectionTimeout);
// ZkSerializer：序列化方式，实现类有：SerializableSerializer、BytesPushThroughSerializer
public ZkClient(String zkServers, int sessionTimeout, int connectionTimeout, ZkSerializer zkSerializer);

• 节点创建 API

// path：节点全路径
public void createPersistent(String path);
// createParents：是否创建父节点
public void createPersistent(String path, boolean createParents);
// acl：权限列表
public void createPersistent(String path, boolean createParents, List<ACL> acl);
// data：节点数据
public void createPersistent(String path, Object data);
public void createPersistent(String path, Object data, List<ACL> acl);
// 创建持久顺序节点
public String createPersistentSequential(String path, Object data);
public String createPersistentSequential(String path, Object data, List<ACL> acl);
// 创建临时节点
public void createEphemeral(String path);
public void createEphemeral(String path, List<ACL> acl);
// mode：节点类型
public String create(String path, Object data, CreateMode mode);
public String create(final String path, Object data, final List<ACL> acl, final CreateMode mode);
public void createEphemeral(String path, Object data);
public void createEphemeral(String path, Object data, List<ACL> acl);
// 创建临时顺序节点
public String createEphemeralSequential(String path, Object data);
public String createEphemeralSequential(String path, Object data, List<ACL> acl);

• 查询节点 API

// 获取子节点列表
public List<String> getChildren(String path);
// watch：是否开启观察
protected List<String> getChildren(final String path, final boolean watch);
// 获取子节点数量
public int countChildren(String path);
// 节点是否存在
protected boolean exists(final String path, final boolean watch);
public boolean exists(String path);

• 删除节点 API

public boolean delete(String path);
// version：节点版本
public boolean delete(final String path, final int version);

• 读写节点数据 API

public <T> T readData(String path);
public <T> T readData(String path, boolean returnNullIfPathNotExists);
public <T> T readData(String path, Stat stat);
protected <T> T readData(final String path, final Stat stat, final boolean watch);
public void writeData(String path, Object object);
public void writeData(String path, Object datat, int expectedVersion);

• 观察节点 API

// 观察节点的数据变化
public void watchForData(final String path);
// 观察子节点的变化
public List<String> watchForChilds(final String path)

4. 总结

本节我们学习了如何使用 ZkClient 的 API 对 Zookeeper 服务节点的操作，还介绍了一些常用的 API。以下是本节内容的总结：

1. 使用 Spring Boot 集成 ZkClient 。
2. ZkClient 的 API。