1.3 Zookeeper 核心概念

文档摘要

1.3 Zookeeper 核心概念 1.3 Zookeeper 核心概念 1.3.1 ZNode：数据组织的核心单元 ZNode是Zookeeper中数据组织的最基本单元，类似于文件系统中的节点（目录或文件）。所有存储在Zookeeper中的数据都以ZNode的形式存在，并构成一个层次化的命名空间，这个命名空间非常像标准的文件系统目录结构。概念详解：层次化命名空间： Zookeeper的命名空间由斜杠（）分隔的路径组成，例如、。这种结构使得数据组织和查找非常直观和方便。数据存储：每个ZNode都可以存储少量的数据。Zookeeper并非设计为大数据存储系统，而是专注于存储配置信息、状态信息、元数据等少量但关键的数据。

1.3 Zookeeper 核心概念

1.3.1 ZNode：数据组织的核心单元

ZNode是Zookeeper中数据组织的最基本单元，类似于文件系统中的节点（目录或文件）。所有存储在Zookeeper中的数据都以ZNode的形式存在，并构成一个层次化的命名空间，这个命名空间非常像标准的文件系统目录结构。

概念详解：

层次化命名空间： Zookeeper的命名空间由斜杠（/）分隔的路径组成，例如 /app1/config、/servers/node1。这种结构使得数据组织和查找非常直观和方便。
数据存储： 每个ZNode都可以存储少量的数据。Zookeeper并非设计为大数据存储系统，而是专注于存储配置信息、状态信息、元数据等少量但关键的数据。通常，ZNode存储的数据大小应保持在KB级别，避免存储GB甚至更大的数据。
元数据： 除了存储数据外，每个ZNode还维护着一组元数据，包括：
- czxid (Creation Transaction ID)：创建ZNode的事务ID。
- mzxid (Modification Transaction ID)：最后一次修改ZNode的事务ID。
- pzxid (Parent ZNode Transaction ID)：子节点列表最后一次修改的事务ID。
- ctime (Creation Time)：ZNode创建的时间戳。
- mtime (Modification Time)：ZNode最后一次修改的时间戳。
- dataVersion：ZNode数据版本号，每次数据更新版本号递增。
- aclVersion：ZNode ACL版本号，每次ACL更新版本号递增。
- ephemeralOwner：如果ZNode是临时节点，则记录创建该节点的会话ID；如果是持久节点，则为0。
- dataLength：ZNode数据内容的长度。
- numChildren：子节点的数量。
ZNode类型： Zookeeper支持不同类型的ZNode，以满足不同的使用场景。主要分为以下几种类型：
- 持久 (Persistent) 节点： 这是最常用的节点类型。持久节点一旦创建，除非显式删除，否则会一直存在于Zookeeper中。即使创建该节点的客户端会话结束，持久节点依然存在。
- 临时 (Ephemeral) 节点： 临时节点的生命周期与创建它的客户端会话绑定。当创建临时节点的客户端会话结束（例如，客户端断开连接或会话超时），Zookeeper会自动删除该临时节点。临时节点不允许拥有子节点。临时节点常用于实现会话管理、leader选举等场景。
- 顺序 (Sequential) 节点： 顺序节点可以是持久的或临时的。当创建一个顺序节点时，Zookeeper会在节点名称后追加一个单调递增的数字后缀。例如，如果创建的节点名为 /tasks/task-，Zookeeper可能会创建名为 /tasks/task-0000000001、/tasks/task-0000000002 等节点。顺序节点常用于实现分布式锁、队列等场景，利用其自动递增的特性来保证操作的顺序性。

代码实践（Java - 使用 Curator Framework）：

Curator Framework是一个流行的Zookeeper客户端库，提供了更高级别的API，简化了Zookeeper的操作。以下代码示例演示了如何使用Curator创建不同类型的ZNode。


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.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.ZooDefs;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
public class ZNodeExample {
    public static void main(String[] args) throws Exception {
        String connectString = "localhost:2181"; // Zookeeper连接地址
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3); // 重试策略
        CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);
        client.start(); // 启动客户端
        try {
            // 1. 创建持久节点
            String persistentPath = "/myPersistentNode";
            client.create()
                    .creatingParentsIfNeeded() // 如果父节点不存在则创建
                    .withMode(CreateMode.PERSISTENT)
                    .forPath(persistentPath, "Persistent Data".getBytes(StandardCharsets.UTF_8));
            System.out.println("Created persistent node: " + persistentPath);
            // 2. 创建临时节点
            String ephemeralPath = "/myEphemeralNode";
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(ephemeralPath, "Ephemeral Data".getBytes(StandardCharsets.UTF_8));
            System.out.println("Created ephemeral node: " + ephemeralPath);
            // 3. 创建顺序持久节点
            String sequentialPersistentPath = "/mySequentialPersistentNode-";
            String createdSequentialPersistentPath = client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.PERSISTENT_SEQUENTIAL)
                    .forPath(sequentialPersistentPath, "Sequential Persistent Data".getBytes(StandardCharsets.UTF_8));
            System.out.println("Created sequential persistent node: " + createdSequentialPersistentPath);
            // 4. 创建顺序临时节点
            String sequentialEphemeralPath = "/mySequentialEphemeralNode-";
            String createdSequentialEphemeralPath = client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                    .forPath(sequentialEphemeralPath, "Sequential Ephemeral Data".getBytes(StandardCharsets.UTF_8));
            System.out.println("Created sequential ephemeral node: " + createdSequentialEphemeralPath);
            // 读取节点数据
            byte[] data = client.getData().forPath(persistentPath);
            System.out.println("Data of " + persistentPath + ": " + new String(data, StandardCharsets.UTF_8));
        } finally {
            client.close(); // 关闭客户端
        }
    }
}

代码详解：

CuratorFrameworkFactory.newClient(connectString, retryPolicy): 创建Curator客户端实例，connectString指定Zookeeper集群地址，retryPolicy定义重试策略，用于处理连接失败等情况。
client.start(): 启动Curator客户端，建立与Zookeeper集群的连接。
client.create()...forPath(path, data): 使用create()接口创建ZNode。
- creatingParentsIfNeeded(): 如果父节点不存在，则自动创建父节点。
- withMode(CreateMode.XXX): 设置ZNode的类型，例如 PERSISTENT、EPHEMERAL、PERSISTENT_SEQUENTIAL、EPHEMERAL_SEQUENTIAL。
- forPath(path, data): 指定ZNode的路径和要存储的数据（byte数组）。
client.getData().forPath(path): 读取指定ZNode路径的数据。
client.close(): 关闭Curator客户端，断开与Zookeeper集群的连接。

Mermaid Graph TD 图：ZNode 结构

图表解释：

Zookeeper Namespace 代表Zookeeper的命名空间。
/ 是根节点。
/app1 和 /servers 是根节点的子节点，可以理解为目录。
/app1/config、/servers/node1、/servers/node2 是更深层次的节点，可以是配置文件或服务器节点信息。
每个方框代表一个ZNode，展示了ZNode的层次结构。

1.3.2 数据模型：轻量级数据存储

Zookeeper的数据模型可以概括为“轻量级数据存储”。它不是一个通用的数据库，而是专注于存储和管理少量关键的元数据和配置信息。

概念详解：

小数据量： Zookeeper设计之初就不是为了存储海量数据。每个ZNode节点存储的数据量应保持在KB级别，通常建议不超过1MB。过大的数据量会影响Zookeeper的性能和稳定性。
元数据和配置信息： Zookeeper主要用于存储以下类型的数据：
- 配置信息： 例如，数据库连接字符串、服务地址列表、应用程序配置参数等。
- 状态信息： 例如，分布式系统中各个组件的运行状态、leader选举的状态、任务队列的状态等。
- 控制信息： 例如，分布式锁的持有者信息、分布式队列的控制信息等。
高性能读操作： Zookeeper集群中的每个服务器都保存了完整的数据副本，客户端可以连接到任何一台服务器进行读操作，从而实现高性能的读取。
保证数据一致性： Zookeeper使用ZAB协议（Zookeeper Atomic Broadcast）来保证集群中所有服务器数据的一致性。所有的写操作都会通过Leader服务器进行协调，确保数据变更被可靠地同步到所有Follower服务器。

内容详解：

Zookeeper的数据模型设计理念是“够用就好”。它牺牲了大数据存储能力，换取了高性能、高可用性和强一致性。这种设计非常适合分布式协调场景，在这些场景中，需要快速、可靠地共享和管理少量的关键数据。

与传统数据库相比，Zookeeper的数据模型有以下特点：

非事务性更新（相对于完整ACID事务）： Zookeeper保证单ZNode操作的原子性，但不支持跨多个ZNode的事务。对于复杂的事务性操作，需要在应用层面进行协调。
最终一致性（通过ZAB协议实现）： 虽然Zookeeper努力保证强一致性，但在网络分区等极端情况下，可能会出现短暂的不一致。ZAB协议确保在大多数情况下数据最终会达到一致状态。
更关注可用性和分区容错性（CAP理论中的AP，同时努力保证C）： Zookeeper在CAP理论中更倾向于AP（可用性和分区容错性），但通过ZAB协议，它也尽力保证数据的一致性。在分布式系统中，可用性和分区容错性往往比强一致性更重要，因为系统需要能够在网络不稳定的情况下继续运行。

代码实践（Java - 读取和更新数据）：


import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import java.nio.charset.StandardCharsets;
public class DataModelExample {
    public static void main(String[] args) throws Exception {
        String connectString = "localhost:2181";
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);
        client.start();
        String path = "/myDataNode";
        try {
            // 创建节点 (如果不存在)
            if (client.checkExists().forPath(path) == null) {
                client.create().creatingParentsIfNeeded().forPath(path, "Initial Data".getBytes(StandardCharsets.UTF_8));
                System.out.println("Created node: " + path + " with initial data.");
            }
            // 读取数据
            byte[] data = client.getData().forPath(path);
            System.out.println("Current data of " + path + ": " + new String(data, StandardCharsets.UTF_8));
            // 更新数据
            String newData = "Updated Data";
            client.setData().forPath(path, newData.getBytes(StandardCharsets.UTF_8));
            System.out.println("Updated data of " + path + " to: " + newData);
            // 再次读取数据验证
            byte[] updatedData = client.getData().forPath(path);
            System.out.println("Data of " + path + " after update: " + new String(updatedData, StandardCharsets.UTF_8));
        } finally {
            client.close();
        }
    }
}

代码详解：

client.checkExists().forPath(path): 检查ZNode是否存在。
client.setData().forPath(path, newData.getBytes(StandardCharsets.UTF_8)): 更新指定ZNode路径的数据。

Mermaid Graph TD 图：数据模型特点

图表解释：

Zookeeper Data Model 概括了数据模型的特点。
Small Data Size (KB) 指出数据模型关注小数据量。
Metadata & Configuration 说明数据模型主要存储元数据和配置信息。
High Read Performance 和 Data Consistency (ZAB) 是数据模型的重要特性。
箭头连接表示特性之间的关联关系，例如，小数据量和元数据配置信息导致了高性能读操作和数据一致性的实现。

1.3.3 Watcher：事件监听机制

Watcher是Zookeeper提供的一种事件监听机制，允许客户端注册对ZNode的监听，并在ZNode发生特定事件时接收通知。这是一种一次性触发的机制，一旦Watcher被触发，需要重新注册才能继续监听。

概念详解：

事件类型： Watcher可以监听以下类型的事件：
- NodeCreated： 监听的ZNode被创建。
- NodeDeleted： 监听的ZNode被删除。
- NodeDataChanged： 监听的ZNode数据内容发生变化。
- NodeChildrenChanged： 监听的ZNode的子节点列表发生变化（增加、删除子节点）。
一次性触发： Watcher是一次性触发的。当Watcher监听的事件发生后，Zookeeper服务器会向客户端发送通知，并且该Watcher会被移除。如果客户端需要持续监听，必须在接收到通知后重新注册Watcher。
客户端注册： 客户端可以通过API在指定的ZNode上注册Watcher。注册时可以指定监听的事件类型。
异步通知： Watcher的通知是异步的。当事件发生时，Zookeeper服务器会异步地向客户端发送通知，客户端无需轮询即可及时获取ZNode的变化信息。

内容详解：

Watcher机制是Zookeeper实现分布式协调的关键特性之一。它使得客户端能够及时感知Zookeeper中数据的变化，从而做出相应的反应。例如：

配置管理： 应用程序可以监听配置信息ZNode的变化，当配置更新时，Watcher会通知应用程序重新加载配置，实现动态配置更新。
服务发现： 服务提供者可以将自己的地址信息注册到Zookeeper的某个ZNode下，服务消费者可以监听该ZNode的子节点变化，当有新的服务提供者上线或下线时，消费者可以及时更新服务列表。
分布式锁： Watcher可以用于实现分布式锁的释放通知。当一个客户端持有锁时，其他等待锁的客户端可以监听锁节点的删除事件。当锁被释放（节点被删除）时，等待客户端会收到通知并尝试重新获取锁。

代码实践（Java - 使用 Curator 注册 Watcher）：


import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import java.nio.charset.StandardCharsets;
public class WatcherExample {
    public static void main(String[] args) throws Exception {
        String connectString = "localhost:2181";
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);
        client.start();
        String path = "/myWatchedNode";
        try {
            // 创建节点 (如果不存在)
            if (client.checkExists().forPath(path) == null) {
                client.create().creatingParentsIfNeeded().forPath(path, "Initial Data".getBytes(StandardCharsets.UTF_8));
                System.out.println("Created node: " + path + " for watcher example.");
            }
            // 使用 NodeCache 注册 Watcher (监听数据变化)
            final NodeCache nodeCache = new NodeCache(client, path);
            nodeCache.getListenable().addListener(new NodeCacheListener() {
                @Override
                public void nodeChanged() throws Exception {
                    byte[] data = nodeCache.getCurrentData().getData();
                    System.out.println("Node data changed! New data: " + (data != null ? new String(data, StandardCharsets.UTF_8) : "null"));
                }
            });
            nodeCache.start();
            System.out.println("NodeCache watcher started on: " + path);
            // 模拟数据变化 (在另一个终端或稍后修改该节点数据)
            System.out.println("Waiting for node data changes... Please modify data of " + path + " in Zookeeper.");
            Thread.sleep(60000); // 等待一段时间，以便观察Watcher触发
        } finally {
            client.close();
        }
    }
}

代码详解：

NodeCache: Curator Framework 提供的 NodeCache 类简化了对单个ZNode的数据变化的监听。它在内部维护了ZNode数据的本地缓存，并通过Watcher机制监听ZNode的变化，当ZNode数据发生变化时，NodeCache 会更新本地缓存并触发监听器。
nodeCache.getListenable().addListener(new NodeCacheListener() { ... }): 添加 NodeCacheListener 监听器，当ZNode数据变化时，nodeChanged() 方法会被调用。
nodeCache.start(): 启动 NodeCache，开始监听。

Mermaid Graph TD 图：Watcher 工作流程

图表解释：

Client 代表Zookeeper客户端。
Register Watcher on ZNode 表示客户端在Zookeeper服务器上注册Watcher。
Zookeeper Server 代表Zookeeper服务器。
ZNode Event Occurs 表示被监听的ZNode发生了事件（例如数据变化）。
Zookeeper Server Notifies Client 表示服务器向客户端发送通知。
Client Receives Notification 表示客户端接收到通知。
Watcher Removed 表示Watcher是一次性触发的，触发后被移除。
Re-register Watcher if needed 表示如果需要持续监听，客户端需要重新注册Watcher。

1.3.4 Session：客户端会话管理

Session (会话) 是客户端与Zookeeper集群之间的连接会话。客户端在连接Zookeeper集群时，会创建一个Session，Zookeeper服务器会为该Session分配一个Session ID。Session用于跟踪客户端的状态，并管理临时节点的生命周期。

概念详解：

会话生命周期： Session有生命周期，从客户端成功连接到Zookeeper集群开始，到客户端显式断开连接或会话超时结束。
Session ID： 每个Session都有一个唯一的Session ID，由Zookeeper服务器分配。客户端通过Session ID 与服务器进行交互。
会话超时 (Session Timeout)： 客户端在创建Session时可以设置会话超时时间。Zookeeper服务器会定期向客户端发送心跳包，客户端也需要定期向服务器发送心跳包。如果在会话超时时间内，服务器没有收到客户端的心跳包，则认为客户端会话已过期，Zookeeper服务器会关闭该Session，并删除该Session创建的所有临时节点。
会话状态： Session有不同的状态，例如：
- CONNECTING： 客户端正在尝试连接Zookeeper服务器。
- CONNECTED： 客户端已成功连接到Zookeeper服务器，会话已建立。
- RECONNECTING： 客户端与服务器连接断开，正在尝试重新连接。
- RECONNECTED： 客户端重新连接成功，会话恢复。
- CLOSED： 会话已关闭。
- EXPIRED： 会话已过期。

内容详解：

Session管理是Zookeeper保证可靠性和会话一致性的重要机制。通过Session，Zookeeper能够：

管理临时节点： 临时节点的生命周期与Session绑定。当Session结束时，Zookeeper会自动清理该Session创建的所有临时节点，这对于实现服务注册与发现、leader选举等场景非常重要。
检测客户端存活状态： 通过心跳机制，Zookeeper可以检测客户端的存活状态。如果客户端长时间没有发送心跳，Zookeeper会认为客户端故障，并进行相应的处理（例如，删除临时节点）。
维护会话状态： Session状态的改变会影响客户端的行为。例如，当Session状态变为 CONNECTED 时，客户端才能正常进行Zookeeper操作。当Session状态变为 EXPIRED 时，客户端需要重新创建Session。

代码实践（Java - Session 事件监听）：

Curator Framework 客户端会自动处理 Session 的连接和重连，并通过 ConnectionStateListener 提供会话状态变化的监听。


import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
public class SessionExample {
    public static void main(String[] args) throws Exception {
        String connectString = "localhost:2181";
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);
        // 添加 ConnectionStateListener 监听 Session 状态变化
        client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                System.out.println("Session State changed to: " + newState);
                switch (newState) {
                    case CONNECTED:
                        System.out.println("Successfully connected to Zookeeper.");
                        break;
                    case LOST:
                        System.out.println("Session lost! Possible session timeout or network issue.");
                        // 通常需要处理Session丢失的情况，例如重新注册Watcher，重新获取锁等
                        break;
                    case RECONNECTED:
                        System.out.println("Reconnected to Zookeeper. Session restored.");
                        // 可以在这里进行一些会话恢复后的操作
                        break;
                    case SUSPENDED:
                        System.out.println("Connection suspended. Waiting for reconnection.");
                        break;
                    case READ_ONLY:
                        System.out.println("Connection is read-only.");
                        break;
                    case LOST_DUE_TO_DISCONNECT:
                        System.out.println("Session lost due to disconnect.");
                        break;
                }
            }
        });
        client.start(); // 启动客户端，建立Session
        try {
            Thread.sleep(Long.MAX_VALUE); // 保持客户端运行，监听Session状态变化
        } finally {
            client.close();
        }
    }
}

代码详解：

client.getConnectionStateListenable().addListener(new ConnectionStateListener() { ... }): 注册 ConnectionStateListener 监听器，用于监听Session状态的变化。
stateChanged(CuratorFramework client, ConnectionState newState): 当Session状态发生变化时，stateChanged() 方法会被调用，newState 参数表示新的Session状态。
ConnectionState 枚举： 定义了Session的各种状态，例如 CONNECTED、LOST、RECONNECTED、SUSPENDED 等。

Mermaid Graph TD 图：Session 生命周期

图表解释：

图表展示了Session从客户端启动到结束的各种状态和状态转换。
CONNECTING -> CONNECTED 表示成功建立会话。
CONNECTED -> SUSPENDED -> RECONNECTED -> CONNECTED 表示会话中断和重连恢复的过程。

作者与出处

原作者：灏天文库

来源：灏天文库

原文： https://aiknowledge.cn/collection/223-zookeeper%E5%88%86%E5%B8%83%E5%BC%8F%E5%8D%8F%E8%B0%83%E6%9C%8D%E5%8A%A1%E6%95%99%E7%A8%8B

整理：灏天文库整理

由灏天文库平台收录，内容或由平台用户上传，仅供学习交流