1 重要理论

1.1 数据模型znode

​ 01.png

​ zk 数据存储结构与标准的 Unix 文件系统非常相似，都是在根节点下挂很多子节点。zk 中没有引入传统文件系统中目录与文件的概念，而是使用了称为znode 的数据节点概念。znode 是 zk 中数据的最小单元，每个 znode 上都可以保存数据，同时还可以挂载子节点，形成一个树形化命名空间。

（1）节点类型

持久节点：其会一直保存在 zk 中，直到将其删除为止。
持久顺序节点：
临时节点：其与创建它的会话是绑定的，会话消失，临时节点消失
临时顺序节点

（2）节点状态

cZxid：Created Zxid，表示当前 znode 被创建时的事务 ID
ctime：Created Time，表示当前 znode 被创建的时间
mZxid：Modified Zxid，表示当前 znode 最后一次被修改时的事务 ID
mtime：Modified Time，表示当前 znode 最后一次被修改时的时间
pZxid：表示当前 znode 的子节点列表最后一次被修改时的事务 ID。注意，只能是其子节点列表变更了才会引起 pZxid 的变更，子节点内容的修改不会影响 pZxid。
cversion：Children Version，表示子节点的版本号。该版本号用于充当乐观锁。
dataVersion：表示当前 znode 数据的版本号。该版本号用于充当乐观锁。
aclVersion：表示当前 znode 的权限 ACL 的版本号。该版本号用于充当乐观锁。
ephemeralOwner：若当前 znode 是持久节点，则其值为 0；若为临时节点，则其值为创建该节点的会话的 SessionID。当会话消失后，会根据 SessionID 来查找与该会话相关的临时节点进行删除。
dataLength：当前 znode 中存放的数据的长度。
numChildren：当前 znode 所包含的子节点的个数。

1.2 ACL

（1）ACL 简介

​ ACL 全称为 Access Control List（访问控制列表），是一种细粒度的权限管理策略，可以针对任意用户与组进行细粒度的权限控制。zk 利用 ACL 控制znode 节点的访问权限，如节点数据读写、节点创建、节点删除、读取子节点列表、设置节点权限等。
​ 扩展知识：UGO（User，Group，Other），是粗粒度权限管理策略。

（2）zk 的 ACL 维度

​ Unix/Linux 系统的 ACL 分为两个维度：组与权限。而 Zookeeper 的 ACL 分为三个维度： 授权策略 scheme、授权对象 id、用户权限 permission。
​ 扩展知识：大多数 Unix 已经支持 ACL，Linux 从 2.6 版本开始也支持 ACL 了
Unix/Linux 中的 ACL：子目录/子文件默认继承父目录的 ACL
zk 中的 ACL：子 znode 不会继承父 znode 的 ACL

A、授权策略 scheme

​ 授权策略用于确定权限验证过程中使用的检验策略（简单地说就是，通过什么来验证权限，即一个用户要访问某个 znode，如何验证其身份），在 zk 中最常用的有四种策略。
IP
digest：使用用户名与密码进行验证
world：不验证
super：

B、 授权对象 id

​ 授权对象指的是权限赋予的用户。不同的授权策略具有不同类型的授权对象。下面是各个授权模式对应的授权对象 id。
ip
digest
world：anyone
Super

C、 权限 Permission

​ 权限指的是通过验证的用户可以对znode 执行的操作。共有五种权限，不过 zk 支持自定义权限。
c：create，允许授权对象在当前节点下创建子节点
d：delete
r：read
w：write
a：acl

1.3 Watcher 机制

​ zk 通过 Watcher 机制实现了发布/订阅模式。

（1）watcher 工作原理

02.png

（2）watcher 事件

​ 对于同一个事件类型，在不同的通知状态中代表的含义是不同的。
<table>
<tr>
<th>客户端所处状态</th>
<th>事件类型（常量值）</th>
<th>触发条件</th>
<th>说明</th>
</tr >
<tr >
<td rowspan="5">SyncConnected</td>
<td>None（-1）</td>
<td>客户端与服务器成功建立会话</td>
<td rowspan="5">此时客户端与服务器处于连接状态</td>
</tr>
<tr >
<td>NodeCreated（1）</td>
<td>Watcher 监听的对应数据节点被创建</td>
</tr>
<tr >
<td>NodeDeleted（2）</td>
<td>Watcher 监听的对应数据节点被删除</td>
</tr>
<tr >
<td>NodeDataChanged（3）</td>
<td>Watcher 监听的对应数据节点的数据内容发生变化</td>
</tr>
<tr >
<td>NodeChildrenChanged（4）</td>
<td>Watcher 监听的节点的子节点列表发生变化</td>
</tr>
<tr >
<td>Disconnected（0）</td>
<td>None（-1）</td>
<td>客户端与zk 断开连接</td>
<td>此时客户端与服务器处于连接断开状态</td>
</tr>
<tr >
<td>Expired（-112）</td>
<td>None（-1）</td>
<td>会话失效</td>
<td>此时客户端会话失效， 通常会收到SessionExpiredException异常</td>
</tr>
<tr >
<td>AuthFailed</td>
<td>None（-1）</td>
<td>使用错误的 scheme进行权限检查</td>
<td>通常会收到AuthFailedException 异常</td>
</tr>
</table>

（3）watcher 特性

​ zk 的watcher 机制具有以下几个特性。
一次性：zk 的watcher 机制不适合监听变化非常频繁的场景
串行性：
轻量级：

1.4 会话

会话是 zk 中最重要的概念之一，客户端与服务端之间的任何交互操作都与会话相关。
ZooKeeper 客户端启动时，首先会与 zk 服务器建立一个 TCP 长连接。连接一旦建立，客户端会话的生命周期也就开始了。

（1）会话状态

常见的会话状态有三种：

A、CONNECTING（重要）

​ 连接中。客户端要创建连接，首先会在客户端创建一个zk 对象。客户端会采用轮询方式逐个获取服务器列表中的 zk 的 IP 进行连接尝试，直到连接成功。注意，在轮询之前，首先会将服务器列表打散，然后再进行轮询。

B、 CONNECTED

​ 已连接。

C、 CLOSED

​ 已关闭。若出现会话超时、权限验证失败或客户端主动退出等情况，客户端状态就变为了 CLOSED，注意，此时客户端的 zk 对象就消失了。

（2）会话连接超时管理

​ 当客户端向 zk 发出连接请求后，是如何知道是否连接成功的呢？当 zk 接收到某客户端会话连接后，其会向该客户端发送连接成功 ACK。当客户端接收到 ACK 后，就知道自己已经与 zk 建立了连接。
​ 若 zk 没有收到连接请求，或客户端没有收到 zk 发送的 ACK 怎么办呢？客户端就需要进行等待，直到发生会话连接超时。然后再进行下一次连接尝试。当然，尝试一直连接不上怎么办？这就依赖于连接时设置的超时重试策略了。
​ 会话连接超时是由客户端维护的。

（3）会话空闲超时管理（重要）

​ zk 为每一个客户端都维护着空闲超时管理。一旦空闲超时，服务端就会认为该客户端已丢失，其会将该会话的 SessionId 从服务端清除。这也就是为什么客户端在空闲时需要定时向服务端发送心跳，就是为了维护这个会话长连接的。服务器是通过空闲超时管理来判断会话是否发生中断的。
​ 会话空闲超时管理是由服务端维护的。其采用了一种特殊的方式——分桶策略。

A、基本概念

​ 分桶策略是指，将空闲超时时间相近的会话放到同一个桶中来进行管理，以减少管理的复杂度。在检查超时时，只需要检查桶中剩下的会话即可，因为在该桶的时间范围内没有超时的会话已经被移出了桶，而桶中存在的会话就是超时的会话。
​ zk 对于会话空闲的超时管理并非是精确的管理，即并非是一超时马上就执行相关的超时操作。

03.png

B、 分桶依据

​ 现要在计算当前的会话需要存放到哪个会话桶中进行管理。分桶的计算依据为：

ExpirationTime= CurrentTime + SessionTimeout
BucketTime = (ExpirationTime/ExpirationInterval + 1) * ExpirationInterval

​ 从以上公式可知，一个桶的大小为 ExpirationInterval 时间。只要 ExpirationTime 落入到同一个桶中，系统就会对其中的会话超时进行统一管理。

（4）会话连接事件

​ 客户端与服务端的长连接失效后，客户端将进行重连。在重连过程中客户端会产生三种会话连接事件：
连接丢失
会话转移
会话失效

2 客户端命令

2.1 启动客户端

（1）连接本机 zk 服务器

04.png

（2）连接其它 zk 服务器

05.png

2.2 查看子节点-ls

​ 查看根节点及/brokers 节点下所包含的所有子节点列表。

06.png

2.3 创建节点-create

（1）创建永久节点

​ 创建一个名称为 china 的znode，其值为 999。

07.png

（2）创建顺序节点

​ 在/china 节点下创建了顺序子节点 beijing、shanghai、guangzhou，它们的数据内容分别为 bj、sh、gz。

08.png

（3）创建临时节点

​ 临时节点与持久节点的区别，在后面 get 命令中可以看到。

09.png

（4）创建临时顺序节点

10.png

2.4 获取节点信息-get

（1）获取持久节点数据

11.png

（2）获取顺序节点信息

12.png

（3）获取临时节点信息

13.png

2.5 更新节点数据内容-set

​ 更新前：

14.png

​ 更新：

15.png
16.png

2.6 删除节点-delete

17.png

​ 若要删除具有子节点的节点，会报错。

18.png

2.7 ACL 操作

（1）查看权限-getAcl

19.png

（2）设置权限

下面的命令是，首先增加了一个认证用户 zs，密码为 123，然后为/china 节点指定只有
zs 用户才可访问该节点，而访问权限为所有权限。

20.png

3 可视化客户端

​ zk 常见的可视化客户端有两个：ZooView 与 ZooInspector。

3.1 ZooView

​ 解压后直接双击运行 startup.bat 即可。

21.png

3.2 ZooInspector

​ 在解压目录的 build 目录下进入 cmd 窗口，然后通过 jar 命令运行下面的 jar 包。

22.png

4 ZKClient 客户端

4.1 简介

​ ZkClient 是一个开源客户端，在 Zookeeper 原生API 接口的基础上进行了包装，更便于开发人员使用。内部实现了 Session 超时重连，Watcher 反复注册等功能。像 dubbo 等框架对其也进行了集成使用。

4.2 API 介绍

​ 以下 API 方法均是 ZkClient 类中的方法。

（1）创建会话

​ ZkClient 中提供了九个构造器用于创建会话。

23.png

查看这些方法的源码可以看到具体的参数名称，这些参数的意义为：

参数名	意义
zkServers	指定zk 服务器列表，由英文状态逗号分开的 host:port 字符串组成
connectionTimeout	设置连接创建超时时间，单位毫秒。在此时间内无法创建与zk 的连接，则直接放弃连接，并抛出异常
sessionTimeout	设置会话超时时间，单位毫秒
zkSerializer	为会话指定序列化器。zk 节点内容仅支持字节数组（byte[]）类型， 且 zk 不负责序列化。在创建 zkClient 时需要指定所要使用的序列化器，例如 Hessian 或Kryo。默认使用 Java 自带的序列化方式进行对象的序列化。当为会话指定了序列化器后，客户端在进行读写操作时就会自动进行序列化与反序列化。
connection	IZkConnection 接口对象，是对zk 原生 API 的最直接包装，是和zk 最直接的交互层，包含了增删改查等一系列方法。该接口最常用的实现类是 zkClient 默认的实现类 ZkConnection，其可以完成绝大部分的业务需求。
operationRetryTimeout	设置重试超时时间，单位毫秒

（2）创建节点

​ ZkClient 中提供了 15 个方法用于创建节点。

24.png

​ 查看这些方法的源码可以看到具体的参数名称，这些参数的意义为：

参数名	意义
path	要创建的节点完整路径
data	节点的初始数据内容，可以传入 Object 类型及 null。zk 原生API中只允许向节点传入 byte[]数据作为数据内容，但 zkClient 中具有自定义序列化器，所以可以传入各种类型对象。
mode	节点类型，CreateMode 枚举常量，常用的有四种类型。 PERSISTENT：持久型 PERSISTENT_SEQUENTIAL：持久顺序型 EPHEMERAL：临时型 EPHEMERAL_SEQUENTIAL：临时顺序型
acl	节点的 ACL 策略
callback	回调接口
context	执行回调时可以使用的上下文对象
createParents	是否级递归创建节点。zk 原生 API 中要创建的节点路径必须存在， 即要创建子节点，父节点必须存在。但zkClient 解决了这个问题， 可以做递归节点创建。没有父节点，可以先自动创建了父节点，然后再在其下创建子节点。

（3）删除节点

​ ZkClient 中提供了 3 个方法用于创建节点。

25.png

​ 查看这些方法的源码可以看到具体的参数名称，这些参数的意义为：

参数名	意义
path	要删除的节点的完整路径
version	要删除的节点中包含的数据版本

（4）更新数据

​ ZkClient 中提供了 3 个方法用于修改节点数据内容。

26.png

​ 查看这些方法的源码可以看到具体的参数名称，这些参数的意义为：

参数名	意义
path	要更新的节点的完整路径
data	要采用的新的数据值
expectedVersion	数据更新后要采用的数据版本号

（5）检测节点是否存在

​ ZkClient 中提供了 2 个方法用于判断指定节点的存在性，但 public 方法就一个：只有一个参数的exists()方法。

27.png

​ 查看这些方法的源码可以看到具体的参数名称，这些参数的意义为：

参数名	意义
path	要判断存在性节点的完整路径
watch	要判断存在性节点及其子孙节点是否具有 watcher 监听

（6）获取节点数据内容

​ ZkClient 中提供了 4 个方法用于获取节点数据内容，但 public 方法就三个。

28.png

​ 查看这些方法的源码可以看到具体的参数名称，这些参数的意义为：

参数名	意义
path	要读取数据内容的节点的完整路径
watch	指定节点及其子孙节点是否具有 watcher 监听
returnNullIfPathNotExists	这是个 boolean 值。默认情况下若指定的节点不存在，则会抛出 KeeperException$NoNodeException 异常。设置该值为 true，若指定节点不存在，则直接返回 null 而不再抛出异常。
stat	指定当前节点的状态信息。不过，执行过后该 stat 值会被最新获取到的 stat 值给替换。

（7）获取子节点列表

​ ZkClient 中提供了 2 个方法用于获取节点的子节点列表，但 public 方法就一个：只有一个参数的 getChildren()方法。

29.png

​ 查看这些方法的源码可以看到具体的参数名称，这些参数的意义为：

参数名	意义
path	要获取子节点列表的节点的完整路径
watch	要获取子节点列表的节点及其子孙节点是否具有 watcher 监听

（8）watcher 注册

​ ZkClient 采用 Listener 来实现 Watcher 监听。客户端可以通过注册相关监听器来实现对zk 服务端事件的订阅。
​ 可以通过 subscribeXxx()方法实现 watcher 注册，即相关事件订阅；通过 unsubscribeXxx()方法取消相关事件的订阅。

30.png

​ 查看这些方法的源码可以看到具体的参数名称，这些参数的意义为：

参数名	意义
path	要操作节点的完整路径
watch	要判断存在性节点及其子孙节点是否具有 watcher 监听
IZkChildListener	子节点数量变化监听器
IZkDataListener	数据内容变化监听器
IZkStateListener	客户端与zk 的会话连接状态变化监听器，可以监听新会话的创建、会话创建出错、连接状态改变。连接状态是系统定义好的枚举类型 Event.KeeperState 的常量

4.3 代码演示

（1）创建工程

创建一个 Maven 的 Java 工程，并导入以下依赖。

31.png

这里仅创建一个 ZkClient 的测试类即可。本例不适合使用 JUnit 测试。

32.png

（2）代码

public class ZKClientTest {
    // 指定 zk 集群 
    private static final String CLUSTER = "zkOS:2181";
    // 指定节点名称 
    private static final String PATH = "/mylog";
    public static void main(String[] args) {
        // ---------------- 创建会话 -----------
        // 创建 zkClient
        ZkClient zkClient = new ZkClient(CLUSTER);
        // 为 zkClient 指定序列化器 
        zkClient.setZkSerializer(new SerializableSerializer()); 
        // ---------------- 创建节点 -----------
        // 指定创建持久节点 
        CreateMode mode = CreateMode.PERSISTENT;
        // 指定节点数据内容 
        String data = "first log";
        // 创建节点 
        String nodeName = zkClient.create(PATH, data, mode); 
        System.out.println("新创建的节点名称为：" + nodeName);
        // ---------------- 获取数据内容 -----------
        Object readData = zkClient.readData(PATH); 
        System.out.println("节点的数据内容为：" + readData);
        // ---------------- 注册 watcher -----------
        zkClient.subscribeDataChanges(PATH, new IZkDataListener() { 
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception{
                System.out.print(" 节 点 " + dataPath); 
                System.out.println("的数据已经更新为了" + data);
            }
            @Override
            public void handleDataDeleted(String dataPath) throws Exception { 
                System.out.println(dataPath + "的数据内容被删除");
            }
        });
        // ---------------- 更新数据内容 -----------
        zkClient.writeData(PATH, "second log"); 
        String updatedData = zkClient.readData(PATH);
        System.out.println("更新过的数据内容为：" + updatedData);

        // ---------------- 删除节点 -----------
        zkClient.delete(PATH);

        // ---------------- 判断节点存在性 -----------
        boolean isExists = zkClient.exists(PATH); 
        System.out.println(PATH + "节点仍存在吗？" + isExists);
    }
}

5 Curator 客户端

5.1 简介

​ Curator 是Netflix 公司开源的一套 zk 客户端框架，与 ZkClient 一样，其也封装了 zk 原生API。其目前已经成为Apache 的顶级项目。同时，Curator 还提供了一套易用性、可读性更强的 Fluent 风格的客户端 API 框架。

5.2 API 介绍

​ 这里主要以 Fluent 风格客户端 API 为主进行介绍。

（1）创建会话

A、普通 API 创建 newClient()

在 CuratorFrameworkFactory 类中提供了两个静态方法用于完成会话的创建。

33.png

查看这些方法的源码可以看到具体的参数名称，这些参数的意义为：

参数名	意义
connectString	指定zk 服务器列表，由英文状态逗号分开的 host:port 字符串组成
sessionTimeoutMs	设置会话超时时间，单位毫秒，默认 60 秒
connectionTimeoutMs	设置连接超时时间，单位毫秒，默认 15 秒
retryPolicy	重试策略，内置有四种策略，分别由以下四个类的实例指定：ExponentialBackoffRetry、RetryNTimes、RetryOneTime、RetryUntilElapsed

B、 Fluent 风格创建

34.png

（2）创建节点 create()

​ 下面以满足各种需求的举例方式分别讲解节点创建的方法。
​ 说明：下面所使用的 client 为前面所创建的Curator 客户端实例。
创建一个节点，初始内容为空
​ 语句：client.create().forPath(path);
​ 说明：默认创建的是持久节点，数据内容为空。
创建一个节点，附带初始内容
​ 语句：client.create().forPath(path, “mydata”.getBytes());
​ 说明：Curator 在指定数据内容时，只能使用 byte[]作为方法参数。
创建一个临时节点，初始内容为空
​ 语句：client.create().withMode(CreateMode.EPHEMERAL).forPath(path);
​ 说明：CreateMode 为枚举类型。
创建一个临时节点，并自动递归创建父节点
​ 语句：client.create().createingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
​ 说明：若指定的节点多级父节点均不存在，则会自动创建。

（3）删除节点delete()

删除一个节点
语句：client.delete().forPath(path);
说明：只能将叶子节点删除，其父节点不会被删除。
删除一个节点，并递归删除其所有子节点
语句：client.delete().deletingChildrenIfNeeded().forPath(path);
说明：该方法在使用时需谨慎。

（4）更新数据setData()

设置一个节点的数据内容
语句：client.setData().forPath(path, newData);
说明：该方法具有返回值，返回值为 Stat 状态对象。

（5）检测节点是否存在 checkExits()

设置一个节点的数据内容
语句：Stat stat = client.checkExists().forPath(path);
说明：该方法具有返回值，返回值为 Stat 状态对象。若 stat 为 null，说明该节点不存在，否则说明节点是存在的。

（6）获取节点数据内容getData()

读取一个节点的数据内容
语句：byte[] data = client.getDate().forPath(path);
说明：其返回值为byte[]数组。

（7）获取子节点列表 getChildren()

读取一个节点的所有子节点列表
语句：List<String> childrenNames = client.getChildren().forPath(path);
说明：其返回值为byte[]数组。

（8）watcher 注册 usingWatcher()

​ curator 中绑定 watcher 的操作有三个：checkExists()、getData()、getChildren()。这三个方法的共性是，它们都是用于获取的。这三个操作用于 watcher 注册的方法是相同的，都是usingWatcher()方法。

35.png

​ 这两个方法中的参数 CuratorWatcher 与 Watcher 都为接口。这两个接口中均包含一个process()方法，它们的区别是，CuratorWatcher 中的 process()方法能够抛出异常，这样的话， 该异常就可以被记录到日志中。

监听节点的存在性变化

Stat stat = client.checkExists().usingWatcher((CuratorWatcher) event -> { System.out.println("节点存在性发生变化");}).forPath(path);

监听节点的内容变化

byte[] data = client.getData().usingWatcher((CuratorWatcher) event -> { System.out.println("节点数据内容发生变化");}).forPath(path);

监听节点子节点列表变化

List<String> sons = client.getChildren().usingWatcher((CuratorWatcher) event -> { System.out.println("节点的子节点列表发生变化");}).forPath(path);

5.3 代码演示

（1）创建工程

​ 创建一个 Maven 的 Java 工程，并导入以下依赖。

36.png

（2）代码

public class FluentTest {
    public static void main(String[] args) throws Exception {
        // ---------------- 创建会话 -----------
        // 创建重试策略对象：第 1 秒重试 1 次，最多重试 3 次 
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // 创建客户端 
        CuratorFramework client = CuratorFrameworkFactory
                                    .builder()
                                    .connectString("zkOS:2181")
                                    .sessionTimeoutMs(15000)
                                    .connectionTimeoutMs(13000)
                                    .retryPolicy(retryPolicy)
                                    .namespace("logs")
                                    .build();
         // 开启客户端 
        client.start();

        // 指定要创建和操作的节点，注意，其是相对于/logs 节点的 
        String nodePath = "/host";

        // ---------------- 创建节点 -----------
        String nodeName = client.create().forPath(nodePath, "myhost".getBytes());
        System.out.println("新创建的节点名称为：" + nodeName);

        // ---------------- 获取数据内容并注册 watcher -----------
        byte[] data = client.getData().usingWatcher(
            (CuratorWatcher) event -> 
            {System.out.println(event.getPath() + "数据内容发生变化");}
        ).forPath(nodePath);
        System.out.println("节点的数据内容为：" + new String(data));

        // ---------------- 更新数据内容 -----------
        client.setData().forPath(nodePath, "newhost".getBytes());
        // 获取更新过的数据内容 
        byte[] newData = client.getData().forPath(nodePath); 
        System.out.println("更新过的数据内容为：" + new String(newData));

        // ---------------- 删除节点 -----------
        client.delete().forPath(nodePath);

        // ---------------- 判断节点存在性 -----------
        Stat stat = client.checkExists().forPath(nodePath);
        boolean isExists = true; 
        if(stat == null) {
            isExists = false;
        }
        System.out.println(nodePath + "节点仍存在吗？" + isExists);
    }
}