JAVA：使用 Curator 进行 ZooKeeper 操作的技术指南

1、简述

Apache Curator 是一个基于 ZooKeeper 的 Java 客户端库，它极大地简化了使用 ZooKeeper 的开发工作。Curator 提供了高层次的 API，封装了很多复杂的 ZooKeeper 操作，例如连接管理、分布式锁、Leader 选举等。

在分布式系统中，ZooKeeper 通常被用来作为协调服务，而 Curator 则为我们提供了更简洁易用的接口，减少了开发的复杂性。本文将介绍 Curator 的核心功能及实践样例。

在这里插入图片描述

2、核心功能

Apache Curator是一个比较完善的ZooKeeper客户端框架，通过封装的一套高级API 简化了ZooKeeper的操作。Curator主要解决了三类问题：

封装ZooKeeper client与ZooKeeper server之间的连接处理
提供了一套Fluent风格的操作API
提供ZooKeeper各种应用场景(recipe，比如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等)的抽象封装

Curator 提供了以下核心组件：

2.1 CuratorFramework

CuratorFramework 是 Curator 的核心类，用于与 ZooKeeper 服务交互。

2.2 Recipes

Curator 提供了多种常见分布式模式的实现，包括：

分布式锁 (InterProcessMutex)
分布式队列 (DistributedQueue)
Leader 选举 (LeaderSelector)
节点缓存 (NodeCache)
路径缓存 (PathChildrenCache)
树缓存 (TreeCache)

3、示例实践

Curator中提供了Zookeeper各种应用场景（Recipe，如共享锁服务、Master选举机制和分布式计算器等）的抽象封装。

3.1 依赖引入

在使用 Curator 前，需要在项目中引入相关的依赖：

<!-- zookeeper支持 -->
<dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.6.4</version>
</dependency>
<!-- curator-recipes -->
<dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>5.5.0</version>
</dependency>
<!-- curator-framework -->
<dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>5.5.0</version>
</dependency>

3.2 初始化 CuratorFramework

以下代码展示了如何初始化 CuratorFramework：

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;public class CuratorExample {public static void main(String[] args) {// 创建 CuratorFramework 实例CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181") // ZooKeeper 地址.sessionTimeoutMs(5000).connectionTimeoutMs(3000).retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();// 启动客户端client.start();System.out.println("CuratorFramework 已启动");// 关闭客户端client.close();}
}

3.3 分布式锁

分布式锁是分布式系统中的一个重要功能，用于协调多进程/线程间的访问。

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 {public static void main(String[] args) throws Exception {// 初始化 CuratorFrameworkCuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();client.start();// 创建分布式锁InterProcessMutex lock = new InterProcessMutex(client, "/distributed-lock");// 尝试获取锁if (lock.acquire(10, TimeUnit.SECONDS)) {try {System.out.println("成功获取锁，执行任务...");Thread.sleep(5000); // 模拟任务} finally {lock.release();System.out.println("锁已释放");}} else {System.out.println("未能获取锁");}client.close();}
}

3.4 Leader 选举

Curator 的 LeaderSelector 提供了简单易用的 Leader 选举功能。

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.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;public class LeaderElectionExample {public static void main(String[] args) throws InterruptedException {CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();client.start();// 创建 LeaderSelectorLeaderSelector leaderSelector = new LeaderSelector(client, "/leader-election", new LeaderSelectorListenerAdapter() {@Overridepublic void takeLeadership(CuratorFramework client) throws Exception {System.out.println("成为 Leader，执行任务...");Thread.sleep(3000); // 模拟任务System.out.println("任务完成，释放 Leader 权限");}});leaderSelector.autoRequeue(); // 自动重新排队参与选举leaderSelector.start();Thread.sleep(Integer.MAX_VALUE); // 保持主线程运行client.close();}
}

3.5 节点缓存

NodeCache 用于监听特定节点的数据变更。

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.retry.ExponentialBackoffRetry;public class NodeCacheExample {public static void main(String[] args) throws Exception {CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();client.start();// 创建 NodeCacheNodeCache nodeCache = new NodeCache(client, "/test-node");nodeCache.getListenable().addListener(() -> {System.out.println("节点数据变更，新的数据为：" + new String(nodeCache.getCurrentData().getData()));});nodeCache.start();// 创建节点并修改数据client.create().orSetData().forPath("/test-node", "initial-data".getBytes());Thread.sleep(1000);client.setData().forPath("/test-node", "updated-data".getBytes());Thread.sleep(5000); // 保持运行观察结果client.close();}
}