Zookeeper

ZooKeeper 详解

1. ZooKeeper 概述

在分布式系统日益普及的今天，如何有效地管理和协调集群中的各个节点成为了一个至关重要的问题。ZooKeeper 正是为了解决这类问题而诞生的，它是一个开源的、分布式的协调服务，旨在为分布式应用提供一致性服务。你可以将其理解为分布式系统的“协调器”或“配置中心”，它帮助开发者构建更健壮、更可靠的分布式应用。

1.1 ZooKeeper 的定位与作用

ZooKeeper 的核心目标是简化分布式协调的复杂性。在没有 ZooKeeper 的情况下，开发者需要自行处理诸如配置管理、领导选举、分布式锁、服务注册与发现等一系列分布式系统中的常见问题。这些问题往往复杂且容易出错，而 ZooKeeper 的出现，将这些通用的协调任务抽象出来，以简单易用的 API 提供给开发者，大大降低了分布式系统开发的难度。

核心作用概括：

• 配置管理: 集中管理和动态更新分布式应用程序的配置信息。

• 领导选举: 在分布式环境中选举出一个主节点，负责协调和管理集群。

• 分布式锁: 提供分布式环境下的互斥锁和共享锁，保证数据一致性。

• 服务注册与发现: 服务提供者注册服务信息，服务消费者发现可用服务实例。

• 分布式队列: 实现分布式环境下的队列，用于任务调度和消息传递。

1.2 ZooKeeper 的关键特性

ZooKeeper 之所以能够成为分布式协调服务的首选方案，得益于其以下几个关键特性：

• 简单的数据模型 (ZNode): ZooKeeper 采用类似于文件系统的树形结构来组织数据，每个节点称为 ZNode。ZNode 可以存储少量数据，并可以设置访问控制列表 (ACL)。这种简洁的数据模型使得 ZooKeeper 的使用非常直观和方便。

• 高性能: ZooKeeper 专为高性能读操作优化。读操作直接从本地服务器完成，而写操作则需要通过 Leader 进行协调，但仍然保持了较高的吞吐量和低延迟。这使得 ZooKeeper 非常适合用于配置读取、服务发现等读多写少的场景。

• 高可靠性: ZooKeeper 集群通常由多个服务器组成，形成一个“ZooKeeper Ensemble”。即使部分服务器发生故障，整个服务仍然可以正常运行。ZooKeeper 使用 Zab (ZooKeeper Atomic Broadcast) 协议来保证数据在集群中的一致性，确保数据的高可用性和可靠性。

• 强一致性 (原子性): ZooKeeper 保证数据在集群中节点之间的一致性。客户端无论连接到哪个 ZooKeeper 服务器，都能看到相同的数据视图。ZooKeeper 使用原子广播机制确保事务的原子性和顺序性，遵循 CAP 理论 中的 CP (Consistency and Partition Tolerance) 原则，牺牲了一定的可用性来保证数据的一致性。

• 顺序访问: ZooKeeper 保证客户端的操作请求按照发送的顺序执行。这对于实现分布式锁、领导选举等场景至关重要。

• 实时性: ZooKeeper 通过 Watcher 机制实现了实时的事件通知。客户端可以在 ZNode 上注册 Watcher，当 ZNode 数据发生变化时，ZooKeeper 服务器会立即通知客户端，实现数据的实时同步。

2. ZooKeeper 架构详解

理解 ZooKeeper 的架构是深入掌握其工作原理和有效应用的关键。ZooKeeper 集群由多个服务器组成，形成一个 ZooKeeper Ensemble。

2.1 服务器角色

在 ZooKeeper Ensemble 中，服务器主要分为三种角色：

• Leader: 每个 ZooKeeper 集群在任何时刻都只有一个 Leader 服务器。Leader 负责处理客户端的所有写请求，并协调 Followers 同步数据。Leader 还负责集群内部的决策，例如 Leader 选举。

• Follower: Follower 服务器负责处理客户端的读请求，并将写请求转发给 Leader。Follower 参与 Leader 选举投票，并与 Leader 保持数据同步。当 Leader 宕机时，Follower 会参与新的 Leader 选举。

• Observer (可选): Observer 服务器也是处理客户端的读请求，但不参与 Leader 选举投票，也不参与写操作的投票过程。Observer 的主要作用是扩展集群的读性能，提高读取吞吐量，但不会影响集群的写性能和数据一致性。

2.2 Zab 协议

ZooKeeper 使用 Zab (ZooKeeper Atomic Broadcast) 协议来保证集群中数据的一致性。Zab 协议是一种基于 Paxos 算法改进的原子广播协议，它确保了以下特性：

• 消息的可靠传递: 所有非故障服务器最终都能接收到 Leader 发送的消息。

• 消息的全局有序性: 所有服务器接收到的消息顺序是一致的。

• 消息的原子性: 要么所有服务器都接收并处理消息，要么都不接收。

Zab 协议主要包括两个阶段：

• Leader 选举 (Leader Election): 当集群启动或 Leader 宕机时，需要进行 Leader 选举。Zab 协议采用快速 Leader 选举算法，从 Follower 中选举出一个新的 Leader。

• 原子广播 (Atomic Broadcast): Leader 选举完成后，进入原子广播阶段。Leader 接收客户端的写请求，并将请求转换为事务 Proposal，然后将 Proposal 广播给所有 Follower。Follower 收到 Proposal 后，进行投票，当 Leader 收到过半 Follower 的投票后，Commit 该事务，并将 Commit 消息广播给所有 Follower。Follower 收到 Commit 消息后，提交事务，完成数据同步。

2.3 会话 (Session)

客户端与 ZooKeeper 服务器之间的连接称为会话 (Session)。会话是 ZooKeeper 提供服务的基础，所有的操作都必须在会话的上下文中进行。

• 会话建立: 客户端通过 TCP 连接到 ZooKeeper 服务器，建立会话。会话建立时，客户端会收到一个 Session ID。

• 会话心跳: 客户端定期向服务器发送心跳包，保持会话的有效性。

• 会话超时: 如果在会话超时时间内，服务器没有收到客户端的心跳包，则认为会话失效。会话失效后，客户端在该会话下创建的临时 ZNode 会被自动删除。

• 会话迁移: 如果客户端连接的服务器宕机，客户端可以尝试连接到其他服务器，会话可以迁移到新的服务器，保证服务的连续性。

2.4 Watcher 机制

Watcher 机制是 ZooKeeper 实现实时事件通知的核心机制。客户端可以在 ZNode 上注册 Watcher，当 ZNode 的数据或子节点发生变化时，ZooKeeper 服务器会主动通知客户端。

• Watcher 注册: 客户端在读取 ZNode 数据时，可以同时注册 Watcher。Watcher 注册是 一次性触发 的，即事件发生后，Watcher 会被移除，如果需要持续监听，需要重新注册 Watcher。

• Watcher 类型: Watcher 可以监听 ZNode 的数据变化 (Data Watcher)、子节点变化 (Child Watcher) 和 ZNode 存在状态变化 (Exists Watcher)。

• Watcher 通知: 当 ZNode 发生变化时，ZooKeeper 服务器会向注册了 Watcher 的客户端发送通知事件。通知事件包括事件类型、事件状态和 ZNode 路径等信息。

3. ZooKeeper 数据模型 (ZNode)

ZooKeeper 的数据模型采用树形结构，类似于文件系统。树中的每个节点称为 ZNode (ZooKeeper Node)。ZNode 可以存储少量数据，并可以设置 ACL。

3.1 ZNode 类型

ZNode 主要分为四种类型，根据其生命周期和特性进行区分：

• Persistent (持久节点): 持久节点是最常用的节点类型。一旦创建，持久节点会一直存在于 ZooKeeper 中，即使创建该节点的客户端会话失效，持久节点也不会被删除。除非客户端显式地删除该节点。

• Ephemeral (临时节点): 临时节点的生命周期与创建该节点的客户端会话绑定。当客户端会话失效时，ZooKeeper 会自动删除该客户端创建的所有临时节点。临时节点常用于实现会话管理、领导选举等场景。

• Persistent_Sequential (持久顺序节点): 持久顺序节点继承了持久节点的特性，同时还具有顺序性。当创建持久顺序节点时，ZooKeeper 会在节点名称后面自动追加一个单调递增的数字后缀。例如，如果创建节点 /locks/lock-，ZooKeeper 可能会创建 /locks/lock-0000000001、/locks/lock-0000000002 等节点。顺序节点常用于实现分布式锁、队列等场景。

• Ephemeral_Sequential (临时顺序节点): 临时顺序节点同时具备临时节点和顺序节点的特性。当客户端会话失效时，临时顺序节点会被自动删除。临时顺序节点也常用于实现分布式锁、队列等场景。

3.2 ZNode 属性

每个 ZNode 都包含以下属性：

• Data (数据): ZNode 可以存储少量数据，通常用于存储配置信息、状态信息等。ZooKeeper 限制了 ZNode 存储的数据大小，默认情况下为 1MB。

• ACL (Access Control List，访问控制列表): ACL 用于控制对 ZNode 的访问权限。可以设置不同用户或用户组对 ZNode 的读、写、创建、删除等权限。

• Stat (状态信息): Stat 记录了 ZNode 的元数据信息，例如创建时间、修改时间、数据版本号、ACL 版本号、子节点版本号等。Stat 信息对于实现乐观锁、数据版本控制等功能非常有用。

3.3 ZNode 操作

客户端可以对 ZNode 进行以下操作：

• create (创建): 创建 ZNode，可以指定 ZNode 类型、数据和 ACL。

• getData (获取数据): 获取 ZNode 存储的数据。

• setData (设置数据): 更新 ZNode 存储的数据。

• delete (删除): 删除 ZNode。

• exists (检查存在): 检查 ZNode 是否存在。

• getChildren (获取子节点): 获取 ZNode 的子节点列表。

4. ZooKeeper 代码实践 (Java 客户端 Curator)

为了简化 ZooKeeper 客户端开发，Apache Curator Framework 提供了一套更高级别的 API，封装了 ZooKeeper 的底层操作，并提供了更丰富的功能，例如重试策略、连接管理、分布式锁、领导选举等。

以下代码示例将使用 Curator Framework 来演示 ZNode 的基本操作和 Watcher 机制。

4.1 添加 Curator 依赖

首先，需要在项目中添加 Curator Framework 的依赖。如果使用 Maven，可以在 pom.xml 文件中添加以下依赖：

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>5.2.0</version> <version>  <!--  请使用最新版本 -->
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.2.0</version> <version>  <!--  请使用最新版本 -->
</dependency>
​

4.2 创建 ZooKeeper 客户端

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class ZookeeperClientExample {
    private static final String ZK_ADDRESS = "127.0.0.1:2181"; // ZooKeeper 服务器地址
    public static void main(String[] args) throws Exception {
        // 创建 Curator 客户端
        CuratorFramework client = CuratorFrameworkFactory.newClient(
                ZK_ADDRESS,
                new ExponentialBackoffRetry(1000, 3) // 重试策略：初始间隔 1 秒，重试 3 次
        );
        client.start(); // 启动客户端
        System.out.println("ZooKeeper client started");
        // 执行 ZNode 操作和 Watcher 示例 (见后续代码)
        client.close(); // 关闭客户端
        System.out.println("ZooKeeper client closed");
    }
}
​

代码详解:

• CuratorFrameworkFactory.newClient(): 使用工厂模式创建 Curator 客户端。

• ZK_ADDRESS: 指定 ZooKeeper 服务器的地址和端口。

• ExponentialBackoffRetry: 设置重试策略，当连接失败或操作失败时，Curator 会根据该策略进行重试。ExponentialBackoffRetry(1000, 3) 表示初始重试间隔为 1 秒，最大重试次数为 3 次，每次重试间隔呈指数增长。

• client.start(): 启动 Curator 客户端，建立与 ZooKeeper 服务器的连接。

• client.close(): 关闭 Curator 客户端，断开与 ZooKeeper 服务器的连接。

4.3 ZNode 基本操作示例

        String path = "/my_znode";
        // 1. 创建持久节点
        client.create().creatingParentsIfNeeded().forPath(path, "initial data".getBytes());
        System.out.println("Created persistent node: " + path);
        // 2. 获取节点数据
        byte[] data = client.getData().forPath(path);
        System.out.println("Data of node " + path + ": " + new String(data));
        // 3. 设置节点数据
        client.setData().forPath(path, "updated data".getBytes());
        System.out.println("Set data for node: " + path);
        // 4. 再次获取节点数据
        data = client.getData().forPath(path);
        System.out.println("Updated data of node " + path + ": " + new String(data));
        // 5. 检查节点是否存在
        boolean exists = client.checkExists().forPath(path) != null;
        System.out.println("Node " + path + " exists: " + exists);
        // 6. 获取子节点 (假设没有子节点)
        List<String> children = client.getChildren().forPath("/");
        System.out.println("Children of root node: " + children);
        // 7. 删除节点
        client.delete().forPath(path);
        System.out.println("Deleted node: " + path);
        // 8. 再次检查节点是否存在
        exists = client.checkExists().forPath(path) != null;
        System.out.println("Node " + path + " exists after deletion: " + exists);
​

代码详解:

• client.create().creatingParentsIfNeeded().forPath(path, "initial data".getBytes()): 创建持久节点 /my_znode，并设置初始数据为 "initial data"。creatingParentsIfNeeded() 方法表示如果父节点不存在，则自动创建父节点。

• client.getData().forPath(path): 获取节点 /my_znode 的数据。

• client.setData().forPath(path, "updated data".getBytes()): 设置节点 /my_znode 的数据为 "updated data"。

• client.checkExists().forPath(path) != null: 检查节点 /my_znode 是否存在。

• client.getChildren().forPath("/"): 获取根节点 / 的子节点列表。

• client.delete().forPath(path): 删除节点 /my_znode。

4.4 Watcher 示例 (数据 Watcher)

        String watchPath = "/watch_node";
        // 创建持久节点 (如果不存在)
        if (client.checkExists().forPath(watchPath) == null) {
            client.create().creatingParentsIfNeeded().forPath(watchPath, "initial watch data".getBytes());
        }
        // 注册数据 Watcher
        client.getData().usingWatcher((org.apache.curator.framework.api.CuratorWatcher) event -> {
            System.out.println("Watcher triggered for path: " + event.getPath());
            System.out.println("Event type: " + event.getType());
            System.out.println("Event state: " + event.getState());
            try {
                byte[] updatedData = client.getData().forPath(watchPath);
                System.out.println("Updated data: " + new String(updatedData));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).forPath(watchPath);
        System.out.println("Data Watcher registered for node: " + watchPath);
        // 修改节点数据，触发 Watcher
        client.setData().forPath(watchPath, "changed watch data".getBytes());
        System.out.println("Data changed for node: " + watchPath);
        Thread.sleep(2000); // 等待 Watcher 触发
​

代码详解:

• client.getData().usingWatcher(...): 注册数据 Watcher 到节点 /watch_node。

• (org.apache.curator.framework.api.CuratorWatcher) event -> { ... }: Lambda 表达式定义了 Watcher 的处理逻辑。当节点 /watch_node 的数据发生变化时，该 Watcher 会被触发，并执行 Lambda 表达式中的代码。

• event.getPath(), event.getType(), event.getState(): 从 Watcher 事件中获取路径、事件类型和事件状态等信息。

• client.getData().forPath(watchPath): 在 Watcher 触发后，重新获取节点 /watch_node 的最新数据。

• Thread.sleep(2000): 为了确保 Watcher 有足够的时间被触发和执行，程序休眠 2 秒。

运行示例代码:

运行上述 ZookeeperClientExample.java 代码，你将看到客户端与 ZooKeeper 服务器建立连接，执行 ZNode 的创建、读取、更新、删除等操作，并演示了数据 Watcher 的注册和触发过程。

5. ZooKeeper 典型应用场景

ZooKeeper 在分布式系统中有着广泛的应用，以下列举几个典型的应用场景：

5.1 配置管理

ZooKeeper 可以作为分布式配置中心，集中管理应用程序的配置信息。

• 集中存储配置: 将应用程序的配置信息存储在 ZooKeeper 的 ZNode 中，例如数据库连接信息、日志级别、线程池大小等。

• 动态更新配置: 应用程序启动时从 ZooKeeper 获取配置信息，并注册 Watcher 监听配置节点的变化。当配置信息发生变化时，ZooKeeper 会通知应用程序，应用程序可以动态更新配置，无需重启。

• 配置版本管理: 利用 ZNode 的版本号机制，可以实现配置的版本管理和回滚。

代码示例 (配置管理 - 简易版):

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;
public class ConfigManagementExample {
    private static final String ZK_ADDRESS = "127.0.0.1:2181";
    private static final String CONFIG_PATH = "/app_config/db_url";
    public static void main(String[] args) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.newClient(
                ZK_ADDRESS,
                new ExponentialBackoffRetry(1000, 3)
        );
        client.start();
        // 创建配置节点 (如果不存在)
        if (client.checkExists().forPath(CONFIG_PATH) == null) {
            client.create().creatingParentsIfNeeded().forPath(CONFIG_PATH, "jdbc:mysql://localhost:3306/mydb".getBytes());
        }
        // 创建 NodeCache 监听配置节点变化
        NodeCache nodeCache = new NodeCache(client, CONFIG_PATH);
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                byte[] data = nodeCache.getCurrentData().getData();
                System.out.println("Config updated: " + new String(data));
                // 应用新的配置信息
                applyNewConfig(new String(data));
            }
        });
        nodeCache.start();
        System.out.println("Current config: " + new String(nodeCache.getCurrentData().getData()));
        Thread.sleep(Long.MAX_VALUE); // 保持程序运行，监听配置变化
        nodeCache.close();
        client.close();
    }
    private static void applyNewConfig(String configData) {
        // 在这里应用新的配置信息，例如重新初始化数据库连接池等
        System.out.println("Applying new config: " + configData);
    }
}
​

代码详解:

• NodeCache: Curator 提供的用于监听单个 ZNode 数据变化的 Cache 工具。

• NodeCacheListener: 监听器接口，nodeChanged() 方法在节点数据变化时被调用。

• nodeCache.getCurrentData().getData(): 获取当前节点的最新数据。

• applyNewConfig(): 模拟应用新的配置信息。

5.2 领导选举 (Leader Election)

在分布式系统中，经常需要选举出一个 Leader 节点来协调和管理集群。ZooKeeper 可以实现可靠的 Leader 选举。

• 临时顺序节点: 每个参与选举的节点都在 ZooKeeper 上创建一个临时顺序节点，例如 /election/candidate-0000000001、/election/candidate-0000000002 等。

• 选举规则: 序号最小的节点被选举为 Leader。

• 监听前序节点: 每个节点监听序号比自己小的最大序号的节点 (前序节点) 的删除事件。

• 重新选举: 当 Leader 节点宕机或会话失效时，其临时节点会被删除，监听该节点删除事件的节点会收到通知，并重新进行选举。

代码示例 (领导选举 - Curator Recipes LeaderLatch):

Curator Recipes 提供了 LeaderLatch 工具类，简化了领导选举的实现。

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import java.util.concurrent.TimeUnit;
public class LeaderElectionExample {
    private static final String ZK_ADDRESS = "127.0.0.1:2181";
    private static final String LATCH_PATH = "/leader_latch";
    public static void main(String[] args) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.newClient(
                ZK_ADDRESS,
                new ExponentialBackoffRetry(1000, 3)
        );
        client.start();
        LeaderLatch leaderLatch = new LeaderLatch(client, LATCH_PATH);
        leaderLatch.addListener(new LeaderLatchListener() {
            @Override
            public void isLeader() {
                System.out.println("I am the leader!");
                // 执行 Leader 任务
            }
            @Override
            public void notLeader() {
                System.out.println("I am not the leader.");
                // 执行 Follower 任务
            }
        });
        leaderLatch.start();
        Thread.sleep(TimeUnit.SECONDS.toMillis(10)); // 保持程序运行一段时间
        leaderLatch.close();
        client.close();
    }
}
​

代码详解:

• LeaderLatch: Curator 提供的 LeaderLatch 工具类，用于实现领导选举。

• LeaderLatchListener: 监听器接口，isLeader() 方法在当前实例成为 Leader 时被调用，notLeader() 方法在当前实例不再是 Leader 时被调用。

• leaderLatch.start(): 启动 LeaderLatch，参与领导选举。

• leaderLatch.close(): 关闭 LeaderLatch，退出领导选举。

5.3 分布式锁 (Distributed Lock)

ZooKeeper 可以实现分布式环境下的互斥锁和共享锁，保证数据的一致性。

• 互斥锁 (Exclusive Lock): 同一时刻只允许一个客户端持有锁，用于保护共享资源的独占访问。可以使用临时顺序节点实现互斥锁。

• 共享锁 (Shared Lock): 允许多个客户端同时持有锁，用于实现读多写少的场景下的并发控制。可以使用持久顺序节点和 Watcher 机制实现共享锁。

代码示例 (分布式互斥锁 - Curator Recipes InterProcessMutex):

Curator Recipes 提供了 InterProcessMutex 工具类，简化了分布式互斥锁的实现.

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import java.util.concurrent.TimeUnit;
public class DistributedLockExample {
    private static final String ZK_ADDRESS = "127.0.0.1:2181";
    private static final String LOCK_PATH = "/distributed_lock";
    public static void main(String[] args) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.newClient(
                ZK_ADDRESS,
                new ExponentialBackoffRetry(1000, 3)
        );
        client.start();
        InterProcessMutex lock = new InterProcessMutex(client, LOCK_PATH);
        try {
            if (lock.acquire(10, TimeUnit.SECONDS)) { // 尝试获取锁，最多等待 10 秒
                System.out.println("Acquired lock, processing...");
                Thread.sleep(5000); // 模拟业务处理
            } else {
                System.out.println("Failed to acquire lock.");
            }
        } finally {
            lock.release(); // 释放锁
            System.out.println("Released lock.");
        }
        client.close();
    }
}
​

代码详解:

• InterProcessMutex: Curator 提供的 InterProcessMutex 工具类，用于实现分布式互斥锁。

• lock.acquire(10, TimeUnit.SECONDS): 尝试获取锁，最多等待 10 秒。如果获取成功返回 true，否则返回 false。

• lock.release(): 释放锁。务必在 finally 块中释放锁，确保锁一定会被释放，避免死锁。

5.4 服务注册与发现 (Service Discovery)

在微服务架构中，服务提供者需要将服务信息注册到注册中心，服务消费者需要从注册中心发现可用的服务实例。ZooKeeper 可以作为服务注册中心。