第四章：Zookeeper 应用场景

文档摘要

第四章：Zookeeper 应用场景第四章：Zookeeper 应用场景 4.1 配置管理在分布式系统中，配置管理是一个核心问题。成百上千的服务实例可能需要共享相同的配置信息，例如数据库连接串、消息队列地址、各种开关参数等。如果每个服务实例都维护一份独立的配置，那么配置变更将会变得极其繁琐且容易出错。Zookeeper提供了一种优雅的解决方案来管理分布式配置。原理详解: Zookeeper利用其树形结构的数据模型来存储配置信息。我们可以将配置信息以节点的形式存储在Zookeeper集群中，每个配置项对应一个ZNode。服务实例在启动时从Zookeeper集群中读取配置信息，并将其缓存在本地。为了实现配置的动态更新，Zookeeper提供了Watcher机制。

第四章：Zookeeper 应用场景

4.1 配置管理

在分布式系统中，配置管理是一个核心问题。成百上千的服务实例可能需要共享相同的配置信息，例如数据库连接串、消息队列地址、各种开关参数等。如果每个服务实例都维护一份独立的配置，那么配置变更将会变得极其繁琐且容易出错。Zookeeper提供了一种优雅的解决方案来管理分布式配置。

原理详解:

Zookeeper利用其树形结构的数据模型来存储配置信息。我们可以将配置信息以节点的形式存储在Zookeeper集群中，每个配置项对应一个ZNode。服务实例在启动时从Zookeeper集群中读取配置信息，并将其缓存在本地。

为了实现配置的动态更新，Zookeeper提供了Watcher机制。服务实例在读取配置信息时，可以同时注册一个Watcher。当Zookeeper集群中配置信息发生变更时（例如，配置节点的数据被修改），Zookeeper会通知所有注册了该节点Watcher的服务实例。服务实例接收到通知后，会重新从Zookeeper集群拉取最新的配置信息，并更新本地缓存，从而实现配置的动态更新。

代码实践 (Java):

以下Java代码示例展示了如何使用Zookeeper进行配置管理。我们使用 Curator Framework 作为Zookeeper客户端，简化了Zookeeper的操作。


import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import java.nio.charset.StandardCharsets;
public class ConfigManagement {
    private static final String ZK_ADDRESS = "127.0.0.1:2181";
    private static final String CONFIG_PATH = "/config/app_config";
    public static void main(String[] args) throws Exception {
        CuratorFramework client = createClient();
        client.start();
        // 1. 初始化配置
        if (client.checkExists().forPath(CONFIG_PATH) == null) {
            System.out.println("Initializing configuration...");
            initConfig(client);
        }
        // 2. 读取配置
        String configValue = readConfig(client);
        System.out.println("Current config value: " + configValue);
        // 3. 监听配置变化
        watchConfigChanges(client);
        // 模拟配置变更 (可以通过Zookeeper客户端工具手动修改节点数据)
        // 或者在程序中修改：
        // updateConfig(client, "New Configuration Value");
        System.out.println("Application started, waiting for config changes...");
        Thread.sleep(Long.MAX_VALUE); // 保持程序运行，等待配置变更事件
    }
    private static CuratorFramework createClient() {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        return CuratorFrameworkFactory.newClient(ZK_ADDRESS, retryPolicy);
    }
    private static void initConfig(CuratorFramework client) throws Exception {
        client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(CONFIG_PATH, "Initial Configuration Value".getBytes(StandardCharsets.UTF_8));
    }
    private static String readConfig(CuratorFramework client) throws Exception {
        byte[] data = client.getData().forPath(CONFIG_PATH);
        return new String(data, StandardCharsets.UTF_8);
    }
    private static void updateConfig(CuratorFramework client, String newValue) throws Exception {
        client.setData().forPath(CONFIG_PATH, newValue.getBytes(StandardCharsets.UTF_8));
        System.out.println("Configuration updated to: " + newValue);
    }
    private static void watchConfigChanges(CuratorFramework client) throws Exception {
        NodeCache cache = new NodeCache(client, CONFIG_PATH);
        cache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                String newConfigValue = readConfig(client);
                System.out.println("Configuration changed! New value: " + newConfigValue);
                // 在这里处理配置变更后的逻辑，例如重新加载配置、更新缓存等
            }
        });
        cache.start();
    }
}

代码详解:

createClient(): 创建CuratorFramework客户端实例，连接到Zookeeper集群。使用了 ExponentialBackoffRetry 重试策略，增强了连接的稳定性。
initConfig(): 初始化配置节点。使用 creatingParentsIfNeeded() 确保父节点不存在时会被自动创建。使用 CreateMode.PERSISTENT 创建持久节点，配置信息会一直保存，即使创建者断开连接。
readConfig(): 读取配置节点的数据，并将其转换为字符串。
updateConfig(): 更新配置节点的数据。
watchConfigChanges(): 使用 NodeCache 监听配置节点的变化。NodeCache 会缓存节点的数据，并在节点数据发生变化时触发 nodeChanged() 方法。在 nodeChanged() 方法中，我们重新读取最新的配置信息，并打印到控制台。实际应用中，可以在这里执行配置变更后的逻辑，例如重新加载配置、更新本地缓存等。

总结:

Zookeeper的配置管理方案具有以下优点：

集中管理: 所有服务实例共享同一份配置，方便管理和维护。
动态更新: 配置变更可以实时推送到所有服务实例，无需重启服务。
高可用性: Zookeeper集群本身具有高可用性，保证了配置管理的可靠性。
版本控制: Zookeeper可以记录配置的历史版本，方便回溯和审计。

4.2 服务发现

在微服务架构中，服务发现是至关重要的基础设施。服务提供者需要将自己的网络地址注册到服务注册中心，服务消费者需要从服务注册中心获取服务提供者的地址列表，才能进行服务调用。Zookeeper可以作为高性能的服务注册中心，实现服务的自动注册和发现。

原理详解:

Zookeeper利用其临时节点和Watcher机制来实现服务发现。

服务注册: 服务提供者在启动时，会在Zookeeper集群中创建一个临时节点，节点路径可以代表服务名称，节点数据可以存储服务提供者的网络地址（IP地址和端口号）。临时节点的特点是，当创建该节点的客户端会话失效（例如，服务提供者宕机或网络中断）时，Zookeeper会自动删除该临时节点。
服务发现: 服务消费者在启动时，会从Zookeeper集群中指定的服务节点下获取子节点列表。每个子节点代表一个服务提供者实例，子节点的数据就是服务提供者的网络地址。服务消费者可以缓存这些地址列表，并根据负载均衡策略选择一个服务提供者进行调用。
服务变更通知: 服务消费者可以监听服务节点下的子节点列表变化。当服务提供者上线或下线（临时节点被创建或删除）时，Zookeeper会通知所有监听该服务节点的消费者。消费者接收到通知后，会重新从Zookeeper集群获取最新的服务提供者地址列表，并更新本地缓存，从而实现服务的动态发现。

代码实践 (Java):

以下Java代码示例展示了如何使用Zookeeper实现服务注册和发现。

服务提供者 (ServiceProvider.java):


import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
public class ServiceProvider {
    private static final String ZK_ADDRESS = "127.0.0.1:2181";
    private static final String SERVICE_REGISTRY_PATH = "/service_registry";
    private static final String SERVICE_NAME = "my_service";
    public static void main(String[] args) throws Exception {
        CuratorFramework client = createClient();
        client.start();
        registerService(client);
        System.out.println("Service provider started, press any key to exit.");
        System.in.read(); // 保持程序运行，模拟服务提供者运行状态
        client.close();
    }
    private static CuratorFramework createClient() {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        return CuratorFrameworkFactory.newClient(ZK_ADDRESS, retryPolicy);
    }
    private static void registerService(CuratorFramework client) throws Exception {
        String servicePath = SERVICE_REGISTRY_PATH + "/" + SERVICE_NAME;
        if (client.checkExists().forPath(servicePath) == null) {
            client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(servicePath);
        }
        String instanceAddress = getLocalAddress() + ":8080"; // 假设服务端口为8080
        String instancePath = servicePath + "/" + "instance-"; // 使用顺序临时节点
        client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(instancePath, instanceAddress.getBytes(StandardCharsets.UTF_8));
        System.out.println("Service instance registered at: " + instanceAddress);
    }
    private static String getLocalAddress() throws UnknownHostException {
        return InetAddress.getLocalHost().getHostAddress();
    }
}

服务消费者 (ServiceConsumer.java):


import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import java.nio.charset.StandardCharsets;
import java.util.List;
public class ServiceConsumer {
    private static final String ZK_ADDRESS = "127.0.0.1:2181";
    private static final String SERVICE_REGISTRY_PATH = "/service_registry";
    private static final String SERVICE_NAME = "my_service";
    public static void main(String[] args) throws Exception {
        CuratorFramework client = createClient();
        client.start();
        discoverServices(client);
        System.out.println("Service consumer started, waiting for service changes...");
        Thread.sleep(Long.MAX_VALUE); // 保持程序运行，等待服务变更事件
        client.close();
    }
    private static CuratorFramework createClient() {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        return CuratorFrameworkFactory.newClient(ZK_ADDRESS, retryPolicy);
    }
    private static void discoverServices(CuratorFramework client) throws Exception {
        String servicePath = SERVICE_REGISTRY_PATH + "/" + SERVICE_NAME;
        // 1. 初始化服务列表
        List<String> instances = getServiceInstances(client, servicePath);
        System.out.println("Initial service instances: " + instances);
        // 2. 监听服务变化
        PathChildrenCache cache = new PathChildrenCache(client, servicePath, true); // true: 缓存节点数据
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case CHILD_ADDED:
                        System.out.println("Service instance added: " + new String(event.getData().getData(), StandardCharsets.UTF_8));
                        updateServiceList(client, servicePath);
                        break;
                    case CHILD_REMOVED:
                        System.out.println("Service instance removed: " + new String(event.getData().getData(), StandardCharsets.UTF_8));
                        updateServiceList(client, servicePath);
                        break;
                    case CHILD_UPDATED: // 服务实例地址更新 (理论上服务发现场景下不常用)
                        System.out.println("Service instance updated: " + new String(event.getData().getData(), StandardCharsets.UTF_8));
                        updateServiceList(client, servicePath);
                        break;
                    default:
                        break;
                }
            }
        });
        cache.start();
    }
    private static List<String> getServiceInstances(CuratorFramework client, String servicePath) throws Exception {
        List<String> children = client.getChildren().forPath(servicePath);
        List<String> instances = new java.util.ArrayList<>();
        for (String child : children) {
            byte[] data = client.getData().forPath(servicePath + "/" + child);
            instances.add(new String(data, StandardCharsets.UTF_8));
        }
        return instances;
    }
    private static void updateServiceList(CuratorFramework client, String servicePath) throws Exception {
        List<String> instances = getServiceInstances(client, servicePath);
        System.out.println("Updated service instances: " + instances);
        // 在这里更新本地服务列表缓存，并进行负载均衡等操作
    }
}

代码详解:

ServiceProvider.java:

registerService():
- 创建服务注册根节点 /service_registry 和服务节点 /service_registry/my_service (如果不存在)。
- 获取本地IP地址和端口号，组合成服务实例地址。
- 在服务节点下创建顺序临时节点 /service_registry/my_service/instance-，节点数据为服务实例地址。使用顺序临时节点可以避免节点名称冲突，并方便管理。

ServiceConsumer.java:

discoverServices():
- getServiceInstances(): 从Zookeeper获取指定服务节点下的子节点列表，并读取每个子节点的数据（服务实例地址）。
- PathChildrenCache: 使用 PathChildrenCache 监听服务节点下的子节点变化。PathChildrenCache 会缓存子节点列表及其数据，并在子节点发生增删改事件时触发 childEvent() 方法。
- childEvent(): 在 childEvent() 方法中，根据事件类型 (CHILD_ADDED, CHILD_REMOVED, CHILD_UPDATED) 打印相应的日志，并调用 updateServiceList() 更新本地服务列表。
- updateServiceList(): 重新从Zookeeper获取最新的服务实例列表，并打印到控制台。实际应用中，可以在这里更新本地服务列表缓存，并根据负载均衡策略选择服务实例进行调用。

总结:

Zookeeper的服务发现方案具有以下优点：

动态发现: 服务实例的上线和下线可以被服务消费者实时感知。
高可用性: 依赖于Zookeeper集群的高可用性，服务注册中心本身具有高可用性。
简单易用: 基于Zookeeper的临时节点和Watcher机制，实现简单高效。
解耦: 服务提供者和服务消费者之间通过Zookeeper解耦，无需直接依赖对方。

4.3 Leader 选举

在分布式系统中，为了保证数据的一致性和操作的正确性，经常需要选举一个Leader节点来负责协调和管理整个集群。例如，在分布式数据库中，Leader节点负责处理写操作，Follower节点负责同步数据和处理读操作。Zookeeper可以提供可靠的Leader选举机制。

原理详解:

Zookeeper利用其临时顺序节点来实现Leader选举。

参与选举: 集群中的每个节点都尝试在Zookeeper集群中创建一个临时顺序节点，节点路径可以代表选举的根节点，节点名称可以使用顺序编号。
选举Leader: 创建节点成功后，每个节点都会获取当前根节点下的所有子节点列表。节点会比较自己的节点名称与其他节点的节点名称，节点名称序号最小的节点被选举为Leader。
监听Leader变化: 每个Follower节点会监听序号比自己小的那个节点的删除事件。如果监听的节点被删除（例如，Leader节点宕机或会话失效），则Follower节点会重新参与Leader选举，重复步骤1和步骤2。
Leader心跳: Leader节点需要定期向Zookeeper集群发送心跳，保持会话的有效性，防止临时节点被意外删除。

代码实践 (Java):

以下Java代码示例展示了如何使用Zookeeper实现Leader选举。


import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import java.util.UUID;
public class LeaderElection {
    private static final String ZK_ADDRESS = "127.0.0.1:2181";
    private static final String ELECTION_PATH = "/leader_election";
    public static void main(String[] args) throws Exception {
        CuratorFramework client = createClient();
        client.start();
        String participantId = UUID.randomUUID().toString();
        LeaderLatch leaderLatch = new LeaderLatch(client, ELECTION_PATH, participantId);
        leaderLatch.addListener(new LeaderLatchListener() {
            @Override
            public void isLeader() {
                System.out.println(participantId + " is now the leader.");
                // 执行 Leader 节点的逻辑
            }
            @Override
            public void notLeader() {
                System.out.println(participantId + " is now a follower.");
                // 执行 Follower 节点的逻辑
            }
        });
        leaderLatch.start();
        System.out.println(participantId + " is participating in leader election, press any key to exit.");
        System.in.read(); // 保持程序运行，模拟节点运行状态
        leaderLatch.close();
        client.close();
    }
    private static CuratorFramework createClient() {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        return CuratorFrameworkFactory.newClient(ZK_ADDRESS, retryPolicy);
    }
}

代码详解:

LeaderLatch: Curator Framework 提供了 LeaderLatch 类，简化了Leader选举的实现。
new LeaderLatch(client, ELECTION_PATH, participantId): 创建 LeaderLatch 实例，需要指定 Zookeeper 客户端、选举根节点路径 /leader_election 和参与者ID (可以使用 UUID 生成唯一ID)。
LeaderLatchListener: 注册 LeaderLatchListener 监听器，监听 Leader 状态变化。
- isLeader(): 当当前节点被选举为 Leader 时，会调用 isLeader() 方法。可以在这里执行 Leader 节点的逻辑。
- notLeader(): 当当前节点不再是 Leader (例如，失去 Leader 资格或启动时未被选为 Leader) 时，会调用 notLeader() 方法。可以在这里执行 Follower 节点的逻辑.
leaderLatch.start(): 启动 LeaderLatch，开始参与 Leader 选举。
leaderLatch.close(): 关闭 LeaderLatch，退出 Leader 选举。

总结:

Zookeeper的Leader选举方案具有以下优点：

可靠性: Zookeeper保证了在任何时刻只有一个Leader被选举出来。即使Leader节点宕机，也能快速重新选举出新的Leader。
公平性: 基于临时顺序节点的选举机制，保证了选举的公平性，先到先得。
简单易用: Curator Framework 等客户端库提供了方便的API，简化了Leader选举的实现。
高性能: Zookeeper集群本身具有高性能，保证了Leader选举的效率。

4.4 分布式锁

在分布式系统中，为了保证数据的一致性和避免资源竞争，经常需要使用分布式锁来控制对共享资源的访问。Zookeeper可以提供高性能、高可靠的分布式锁服务。

原理详解:

Zookeeper可以使用临时顺序节点来实现分布式锁。

获取锁: 客户端尝试在Zookeeper集群中创建一个临时顺序节点，节点路径可以代表锁的名称，节点名称可以使用顺序编号。
判断锁持有者: 创建节点成功后，客户端获取当前锁节点下的所有子节点列表。客户端会比较自己的节点名称与其他节点的节点名称，如果自己的节点名称序号是最小的，则认为自己获得了锁。
监听锁释放: 如果客户端没有获得锁（节点名称序号不是最小的），则需要监听序号比自己小的那个节点的删除事件。当监听的节点被删除（例如，锁持有者释放锁或会话失效）时，客户端会重新尝试获取锁，重复步骤2。
释放锁: 锁持有者在完成对共享资源的访问后，需要删除自己创建的临时顺序节点，释放锁。其他等待锁的客户端会收到节点删除事件通知，并重新尝试获取锁。

代码实践 (Java):

以下Java代码示例展示了如何使用Zookeeper实现分布式锁。


import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import java.util.concurrent.TimeUnit;
public class DistributedLock {
    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 client1 = createClient();
        CuratorFramework client2 = createClient();
        client1.start();
        client2.start();
        InterProcessMutex lock1 = new InterProcessMutex(client1, LOCK_PATH);
        InterProcessMutex lock2 = new InterProcessMutex(client2, LOCK_PATH);
        // 客户端1 获取锁
        System.out.println("Client 1 trying to acquire lock...");
        lock1.acquire();
        System.out.println("Client 1 acquired lock.");
        // 客户端2 尝试获取锁，会被阻塞直到客户端1释放锁
        new Thread(() -> {
            try {
                System.out.println("Client 2 trying to acquire lock...");
                lock2.acquire();
                System.out.println("Client 2 acquired lock.");
                System.out.println("Client 2 is working...");
                TimeUnit.SECONDS.sleep(3); // 模拟客户端2持有锁并工作一段时间
                lock2.release();
                System.out.println("Client 2 released lock.");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
        System.out.println("Client 1 is working...");
        TimeUnit.SECONDS.sleep(5); // 模拟客户端1持有锁并工作一段时间
        lock1.release();
        System.out.println("Client 1 released lock.");
        TimeUnit.SECONDS.sleep(10); // 等待客户端2完成
        client1.close();
        client2.close();
    }
    private static CuratorFramework createClient() {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        return CuratorFrameworkFactory.newClient(ZK_ADDRESS, retryPolicy);
    }
}

代码详解:

InterProcessMutex: Curator Framework 提供了 InterProcessMutex 类，简化了分布式互斥锁的实现。
new InterProcessMutex(client, LOCK_PATH): 创建 InterProcessMutex 实例，需要指定 Zookeeper 客户端和锁路径 /distributed_lock。

作者与出处

原作者：灏天文库

来源：灏天文库

原文： 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

整理：灏天文库整理

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