Zookeeper

本文集文档发布于灏天文库

第三章：Zookeeper 客户端 API 使用

文档摘要

第三章：Zookeeper 客户端 API 使用 第三章：Zookeeper 客户端 API 使用 3.1 Zookeeper 客户端 API 概述 Zookeeper 客户端 API 提供了一组接口，允许应用程序连接到 Zookeeper 集群并执行各种操作，例如创建、读取、更新和删除 ZNodes（Zookeeper 中的节点），以及监听 ZNode 的变化事件。 核心功能: 连接管理: 建立和维护与 Zookeeper 集群的连接。 ZNode 操作 (CRUD): 创建 (Create)、读取 (Read)、更新 (Update)、删除 (Delete) ZNodes。 Watcher 机制: 注册监听器 (Watcher) 以接收 ZNode 状态变化通知。

第三章：Zookeeper 客户端 API 使用

第三章：Zookeeper 客户端 API 使用

3.1 Zookeeper 客户端 API 概述

Zookeeper 客户端 API 提供了一组接口，允许应用程序连接到 Zookeeper 集群并执行各种操作，例如创建、读取、更新和删除 ZNodes（Zookeeper 中的节点），以及监听 ZNode 的变化事件。

核心功能:

  • 连接管理: 建立和维护与 Zookeeper 集群的连接。

  • ZNode 操作 (CRUD): 创建 (Create)、读取 (Read)、更新 (Update)、删除 (Delete) ZNodes。

  • Watcher 机制: 注册监听器 (Watcher) 以接收 ZNode 状态变化通知。

  • ACL (访问控制列表): 管理 ZNode 的访问权限。

  • 会话管理: 处理客户端会话的创建、维持和过期。

  • 异步操作: 支持异步 API 调用以提高性能。

常用客户端库:

Zookeeper 官方提供了 Java 和 C 客户端库。此外，还有许多第三方客户端库支持其他编程语言，例如 Python (Kazoo, PyZookeeper)、Go (go-zookeeper)、Node.js (node-zookeeper-client) 等。 本章的代码示例将主要使用 Java 客户端 API 进行演示，因为它最为常用和成熟。

3.2 客户端初始化与连接

要开始使用 Zookeeper 客户端 API，首先需要初始化一个 Zookeeper 客户端实例并连接到 Zookeeper 集群。

Java 客户端代码示例:

import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
public class ZookeeperClientExample {
    private static final String CONNECT_STRING = "127.0.0.1:2181"; // Zookeeper 集群地址
    private static final int SESSION_TIMEOUT = 5000; // 会话超时时间 (毫秒)
    private static CountDownLatch connectedSignal = new CountDownLatch(1); // 用于等待连接成功的信号量
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = connectZookeeper();
        if (zooKeeper != null) {
            System.out.println("Zookeeper 连接成功!");
            // 在这里进行后续的 Zookeeper 操作...
            zooKeeper.close(); // 关闭连接
            System.out.println("Zookeeper 连接已关闭。");
        }
    }
    public static ZooKeeper connectZookeeper() throws IOException, InterruptedException {
        ZooKeeper zooKeeper = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Watcher.Event.KeeperState.SyncConnected) {
                    connectedSignal.countDown(); // 连接成功，信号量减一
                }
            }
        });
        connectedSignal.await(); // 等待连接成功信号
        return zooKeeper;
    }
}
​

代码详解:

  1. 引入依赖: 首先需要引入 Zookeeper 客户端库的依赖。如果您使用 Maven，可以在 pom.xml 文件中添加以下依赖：

    <dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.6.3</version>  <!--  请根据实际情况选择版本 -->
</dependency>
​

  2. CONNECT_STRING: 定义 Zookeeper 集群的连接字符串。可以是单个 Zookeeper 服务器地址 (例如 "127.0.0.1:2181") 或多个服务器地址 (例如 "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183")，用逗号分隔。 客户端会尝试连接列表中的服务器，直到成功为止。

  3. SESSION_TIMEOUT: 设置客户端会话超时时间，单位为毫秒。如果客户端在超时时间内没有向服务器发送心跳，服务器会认为会话已过期并关闭连接。

  4. CountDownLatch: 使用 CountDownLatch 来实现同步等待，确保在连接成功后再进行后续操作。

  5. new ZooKeeper(...): 创建 ZooKeeper 客户端实例，构造函数的参数包括：

    • 连接字符串 (CONNECT_STRING): Zookeeper 集群地址。

    • 会话超时时间 (SESSION_TIMEOUT): 客户端会话超时时间。

    • Watcher: 默认的 Watcher 实例，用于接收来自 Zookeeper 服务器的事件通知。

  6. Watcher 实现: 匿名内部类实现了 Watcher 接口，并重写了 process() 方法。当客户端连接状态发生变化时，process() 方法会被调用。

    • event.getState() == Watcher.Event.KeeperState.SyncConnected: 判断连接状态是否为 SyncConnected，表示客户端已成功连接到 Zookeeper 服务器。

    • connectedSignal.countDown(): 当连接成功时，调用 countDown() 方法使 CountDownLatch 的计数器减一，从而释放 connectedSignal.await() 的阻塞，允许程序继续执行。

  7. connectedSignal.await(): 调用 await() 方法使主线程阻塞，直到 connectedSignal 的计数器变为零（即连接成功）。

  8. zooKeeper.close(): 在完成 Zookeeper 操作后，务必关闭连接以释放资源。

Graph TD 图示连接过程:

3.3 ZNode 的创建 (Create)

create() 方法用于在 Zookeeper 中创建 ZNode。可以指定 ZNode 的路径、数据、ACL 权限和创建模式。

Java 客户端代码示例:

import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
public class ZookeeperCreateNode {
    private static final String CONNECT_STRING = "127.0.0.1:2181";
    private static final int SESSION_TIMEOUT = 5000;
    private static CountDownLatch connectedSignal = new CountDownLatch(1);
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = connectZookeeper();
        if (zooKeeper != null) {
            System.out.println("Zookeeper 连接成功!");
            String path = "/my_node";
            byte[] data = "Hello Zookeeper".getBytes();
            CreateMode createMode = CreateMode.PERSISTENT; // 持久节点
            try {
                String createdPath = zooKeeper.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode);
                System.out.println("ZNode 创建成功，路径: " + createdPath);
            } catch (KeeperException.NodeExistsException e) {
                System.out.println("ZNode 已存在: " + path);
            }
            zooKeeper.close();
        }
    }
    // ... connectZookeeper() 方法与前面示例相同 ...
    public static ZooKeeper connectZookeeper() throws IOException, InterruptedException {
        // ... (代码省略) ...
        return null; // 为了编译通过，实际代码需要返回 ZooKeeper 实例
    }
}
​

代码详解:

  1. zooKeeper.create(path, data, acl, createMode): 调用 create() 方法创建 ZNode。

    • path: 要创建的 ZNode 路径，例如 "/my_node"。

    • data: 要存储在 ZNode 中的数据，以字节数组形式表示。

    • acl: 访问控制列表 (ACL)，用于设置 ZNode 的访问权限。 ZooDefs.Ids.OPEN_ACL_UNSAFE 表示开放的 ACL，允许任何人访问。生产环境中应使用更安全的 ACL 设置。

    • createMode: 创建模式，指定 ZNode 的类型。

  2. CreateMode 枚举: 定义了 ZNode 的创建模式：

    • PERSISTENT (持久节点): ZNode 被创建后，即使创建它的客户端会话结束，ZNode 仍然存在。除非显式删除，否则一直存在。

    • EPHEMERAL (临时节点): ZNode 的生命周期与创建它的客户端会话绑定。当会话结束（例如客户端断开连接或会话超时），ZNode 会被自动删除。

    • PERSISTENT_SEQUENTIAL (持久顺序节点): 与持久节点类似，但 Zookeeper 会在路径末尾自动追加一个单调递增的序列号。例如，如果创建路径 "/my_node" 的持久顺序节点，实际创建的路径可能是 "/my_node0000000001"。

    • EPHEMERAL_SEQUENTIAL (临时顺序节点): 与临时节点类似，但 Zookeeper 会在路径末尾自动追加一个单调递增的序列号。

  3. 异常处理 KeeperException.NodeExistsException: 如果尝试创建的 ZNode 路径已存在，create() 方法会抛出 KeeperException.NodeExistsException 异常。代码中使用了 try-catch 块来捕获并处理这个异常。

Graph TD 图示创建节点过程:

3.4 ZNode 数据的读取 (Get Data)

getData() 方法用于获取 ZNode 中存储的数据。可以注册 Watcher 监听 ZNode 数据变化事件。

Java 客户端代码示例:

import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
public class ZookeeperGetData {
    private static final String CONNECT_STRING = "127.0.0.1:2181";
    private static final int SESSION_TIMEOUT = 5000;
    private static CountDownLatch connectedSignal = new CountDownLatch(1);
    private static CountDownLatch dataChangeSignal = new CountDownLatch(1); // 用于监听数据变化的信号量
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = connectZookeeper();
        if (zooKeeper != null) {
            System.out.println("Zookeeper 连接成功!");
            String path = "/my_node";
            createNodeIfNotExist(zooKeeper, path, "Initial Data".getBytes()); // 确保节点存在
            // 获取数据并注册 Watcher
            byte[] data = zooKeeper.getData(path, new DataWatcher(), null);
            System.out.println("获取数据: " + new String(data));
            System.out.println("等待数据变化...");
            dataChangeSignal.await(); // 等待数据变化事件
            byte[] newData = zooKeeper.getData(path, false, null); // 再次获取数据，不注册 Watcher
            System.out.println("变化后的数据: " + new String(newData));
            zooKeeper.close();
        }
    }
    static class DataWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            if (event.getType() == Event.EventType.NodeDataChanged) {
                System.out.println("数据变化事件触发，路径: " + event.getPath());
                dataChangeSignal.countDown(); // 数据变化，信号量减一
            }
        }
    }
    // 辅助方法：如果节点不存在则创建
    private static void createNodeIfNotExist(ZooKeeper zooKeeper, String path, byte[] data)
            throws KeeperException, InterruptedException {
        try {
            zooKeeper.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        } catch (KeeperException.NodeExistsException e) {
            // 节点已存在，忽略
        }
    }
    // ... connectZookeeper() 方法与前面示例相同 ...
    public static ZooKeeper connectZookeeper() throws IOException, InterruptedException {
        // ... (代码省略) ...
        return null; // 为了编译通过，实际代码需要返回 ZooKeeper 实例
    }
}
​

代码详解:

  1. zooKeeper.getData(path, watcher, stat): 调用 getData() 方法获取 ZNode 数据。

    • path: 要获取数据的 ZNode 路径。

    • watcher: Watcher 实例，用于监听 ZNode 数据变化事件。如果不需要监听，可以传入 null 或 false (在某些客户端库中)。

    • stat: Stat 对象，用于存储 ZNode 的元数据信息，例如版本号、创建时间等。可以传入 null 表示不需要获取元数据。

  2. DataWatcher 类: 实现了 Watcher 接口，用于处理 ZNode 数据变化事件。

    • event.getType() == Event.EventType.NodeDataChanged: 判断事件类型是否为 NodeDataChanged，表示 ZNode 数据发生了变化。

    • dataChangeSignal.countDown(): 当数据变化事件发生时，调用 countDown() 方法释放 dataChangeSignal.await() 的阻塞，使程序继续执行。

  3. createNodeIfNotExist() 辅助方法: 用于确保在执行 getData() 之前，目标 ZNode 已经存在。

Graph TD 图示读取数据过程:

3.5 ZNode 数据的更新 (Set Data)

setData() 方法用于更新 ZNode 中存储的数据。可以指定版本号进行乐观锁控制，防止并发更新冲突。

Java 客户端代码示例:

import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
public class ZookeeperSetData {
    private static final String CONNECT_STRING = "127.0.0.1:2181";
    private static final int SESSION_TIMEOUT = 5000;
    private static CountDownLatch connectedSignal = new CountDownLatch(1);
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = connectZookeeper();
        if (zooKeeper != null) {
            System.out.println("Zookeeper 连接成功!");
            String path = "/my_node";
            createNodeIfNotExist(zooKeeper, path, "Initial Data".getBytes()); // 确保节点存在
            // 获取当前版本号
            Stat stat = new Stat();
            zooKeeper.getData(path, false, stat);
            int version = stat.getVersion();
            System.out.println("当前版本号: " + version);
            // 更新数据，指定版本号
            byte[] newData = "Updated Data".getBytes();
            try {
                Stat newStat = zooKeeper.setData(path, newData, version);
                System.out.println("数据更新成功，新版本号: " + newStat.getVersion());
            } catch (KeeperException.BadVersionException e) {
                System.out.println("版本号冲突，更新失败!");
            }
            zooKeeper.close();
        }
    }
    // ... createNodeIfNotExist() 和 connectZookeeper() 方法与前面示例相同 ...
    private static void createNodeIfNotExist(ZooKeeper zooKeeper, String path, byte[] data)
            throws KeeperException, InterruptedException {
        // ... (代码省略) ...
    }
    public static ZooKeeper connectZookeeper() throws IOException, InterruptedException {
        // ... (代码省略) ...
        return null; // 为了编译通过，实际代码需要返回 ZooKeeper 实例
    }
}
​

代码详解:

  1. zooKeeper.setData(path, data, version): 调用 setData() 方法更新 ZNode 数据。

    • path: 要更新数据的 ZNode 路径。

    • data: 要更新的数据，以字节数组形式表示。

    • version: 期望的版本号。Zookeeper 使用版本号实现乐观锁。

      • 如果传入的版本号与 ZNode 的当前版本号一致，更新操作会成功。

      • 如果传入的版本号为 -1，则表示不进行版本检查，强制更新。

  2. Stat 对象: Stat 对象用于获取 ZNode 的元数据信息，包括版本号。在更新数据之前，先使用 getData() 获取 ZNode 的 Stat 对象，从中获取当前的 version。

  3. 异常处理 KeeperException.BadVersionException: 如果传入的版本号与 ZNode 的当前版本号不一致，setData() 方法会抛出 KeeperException.BadVersionException 异常，表示版本冲突，更新失败。

Graph TD 图示更新数据过程:

3.6 ZNode 的删除 (Delete)

delete() 方法用于删除 ZNode。可以指定版本号进行乐观锁控制。

Java 客户端代码示例:

import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
public class ZookeeperDeleteNode {
    private static final String CONNECT_STRING = "127.0.0.1:2181";
    private static final int SESSION_TIMEOUT = 5000;
    private static CountDownLatch connectedSignal = new CountDownLatch(1);
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = connectZookeeper();
        if (zooKeeper != null) {
            System.out.println("Zookeeper 连接成功!");
            String path = "/my_node";
            createNodeIfNotExist(zooKeeper, path, "Initial Data".getBytes()); // 确保节点存在
            // 获取当前版本号
            Stat stat = new Stat();
            zooKeeper.getData(path, false, stat);
            int version = stat.getVersion();
            System.out.println("当前版本号: " + version);
            // 删除节点，指定版本号
            try {
                zooKeeper.delete(path, version);
                System.out.println("ZNode 删除成功!");
            } catch (KeeperException.BadVersionException e) {
                System.out.println("版本号冲突，删除失败!");
            } catch (KeeperException.NotEmptyException e) {
                System.out.println("ZNode 不是空节点，无法删除!");
            }
            zooKeeper.close();
        }
    }
    // ... createNodeIfNotExist() 和 connectZookeeper() 方法与前面示例相同 ...
    private static void createNodeIfNotExist(ZooKeeper zooKeeper, String path, byte[] data)
            throws KeeperException, InterruptedException {
        // ... (代码省略) ...
    }
    public static ZooKeeper connectZookeeper() throws IOException, InterruptedException {
        // ... (代码省略) ...
        return null; // 为了编译通过，实际代码需要返回 ZooKeeper 实例
    }
}
​

代码详解:

  1. zooKeeper.delete(path, version): 调用 delete() 方法删除 ZNode。

    • path: 要删除的 ZNode 路径。

    • version: 期望的版本号。与 setData() 类似，用于乐观锁控制。

      • 如果传入的版本号与 ZNode 的当前版本号一致，删除操作会成功。

      • 如果传入的版本号为 -1，则表示不进行版本检查，强制删除。

  2. 异常处理:

    • KeeperException.BadVersionException: 版本号冲突，删除失败。

    • KeeperException.NotEmptyException: 尝试删除的 ZNode 不是空节点，无法删除。Zookeeper 默认不允许删除非空节点，需要先删除其子节点。

Graph TD 图示删除节点过程:

3.7 检查 ZNode 是否存在 (Exists)

exists() 方法用于检查 ZNode 是否存在。可以注册 Watcher 监听 ZNode 创建、删除或数据变化事件。

Java 客户端代码示例:

import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
public class ZookeeperExists {
    private static final String CONNECT_STRING = "127.0.0.1:2181";
    private static final int SESSION_TIMEOUT = 5000;
    private static CountDownLatch connectedSignal = new CountDownLatch(1);
    private static CountDownLatch nodeEventSignal = new CountDownLatch(1); // 用于监听节点事件的信号量
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = connectZookeeper();
        if (zooKeeper != null) {
            System.out.println("Zookeeper 连接成功!");
            String path = "/my_node";
            // 检查节点是否存在并注册 Watcher
            Stat stat = zooKeeper.exists(path, new ExistsWatcher());
            if (stat != null) {
                System.out.println("ZNode 存在，版本号: " + stat.getVersion());
            } else {
                System.out.println("ZNode 不存在");
            }
            System.out.println("等待节点事件...");
            nodeEventSignal.await(); // 等待节点事件
            Stat newStat = zooKeeper.exists(path, false); // 再次检查节点是否存在，不注册 Watcher
            if (newStat != null) {
                System.out.println("ZNode 现在存在");
            } else {
                System.out.println("ZNode 现在不存在");
            }
            zooKeeper.close();
        }
    }
    static class ExistsWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            if (event.getType() == Event.EventType.NodeCreated ||
                event.getType() == Event.EventType.NodeDeleted ||
                event.getType() == Event.EventType.NodeDataChanged) {
                System.out.println("节点事件触发，类型: " + event.getType() + ", 路径: " + event.getPath());
                nodeEventSignal.countDown(); // 节点事件发生，信号量减一
            }
        }
    }
    // ... connectZookeeper() 方法与前面示例相同 ...
    public static ZooKeeper connectZookeeper() throws IOException, InterruptedException {
        // ... (代码省略) ...
        return null; // 为了编译通过，实际代码需要返回 ZooKeeper 实例
    }
}
​

代码详解:

  1. zooKeeper.exists(path, watcher): 调用 exists() 方法检查 ZNode 是否存在。

    • path: 要检查的 ZNode 路径。

    • watcher: Watcher 实例，用于监听 ZNode 的创建、删除或数据变化事件。可以传入 null 或 false (在某些客户端库中) 表示不监听。

  2. 返回值:

    • 如果 ZNode 存在，exists() 方法返回 Stat 对象，包含 ZNode 的元数据信息。

    • 如果 ZNode 不存在，exists() 方法返回 null.

  3. ExistsWatcher 类: 实现了 Watcher 接口，用于处理 ZNode 的创建、删除或数据变化事件。

    • event.getType() == Event.EventType.NodeCreated || event.getType() == Event.EventType.NodeDeleted || event.getType() == Event.EventType.NodeDataChanged: 判断事件类型是否为 NodeCreated, NodeDeleted 或 NodeDataChanged，表示节点创建、删除或数据变化事件。

    • nodeEventSignal.countDown(): 当节点事件发生时，调用 countDown() 方法释放 nodeEventSignal.await() 的阻塞，使程序继续执行。

Graph TD 图示检查节点存在过程:

3.8 获取子节点列表 (Get Children)

getChildren() 方法用于获取指定 ZNode 的子节点列表。可以注册 Watcher 监听子节点列表变化事件。

Java 客户端代码示例:

发布者: 作者: 转发
评论区 (0)
U
0/500