基于Apache Curator框架的ZooKeeper基本用法详解

    科技2026-06-21  35

    简介

            Apache Curator是一个比较完善的ZooKeeper客户端框架,通过封装的一套高级API 简化了ZooKeeper的操作。通过查看官方文档,可以发现Curator主要解决了三类问题:

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

    Curator主要从以下几个方面降低了zk使用的复杂性:

    重试机制:提供可插拔的重试机制, 它将给捕获所有可恢复的异常配置一个重试策略,并且内部也提供了几种标准的重试策略(比如指数补偿)连接状态监控:Curator初始化之后会一直对zk连接进行监听,一旦发现连接状态发生变化将会作出相应的处理zk客户端实例管理:Curator会对zk客户端到server集群的连接进行管理,并在需要的时候重建zk实例,保证与zk集群连接的可靠性各种使用场景支持:Curator实现了zk支持的大部分使用场景(甚至包括zk自身不支持的场景),这些实现都遵循了zk的最佳实践,并考虑了各种极端情况

    基于Curator的ZooKeeper基本用法

    首先需要在pom.xml中添加如下依赖:

    <!-- Apache Curator --> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-framework</artifactId> <version>4.0.0</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>4.0.0</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-x-discovery</artifactId> <version>4.0.0</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-test</artifactId> <version>4.0.0</version> <scope>test</scope> </dependency>

    1、初始化以及创建节点

    import org.apache.curator.RetryPolicy; 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.junit.Before; import org.junit.Test; /** * 测试Apache Curator框架的基本用法 * API文档:http://curator.apache.org/apidocs/index.html * */ public class TestApacheCurator { //会话超时时间 private final int SESSION_TIMEOUT = 30 * 1000; //连接超时时间 private final int CONNECTION_TIMEOUT = 3 * 1000; //ZooKeeper服务地址 private static final String SERVER = "192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102"; //创建连接实例 private CuratorFramework client = null; /** * baseSleepTimeMs:初始的重试等待时间 * maxRetries:最多重试次数 */ RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3); @Before public void init(){ //创建 CuratorFrameworkImpl实例 client = CuratorFrameworkFactory.newClient(SERVER, SESSION_TIMEOUT, CONNECTION_TIMEOUT, retryPolicy); //启动 client.start(); } /** * 测试创建节点 * @throws Exception */ @Test public void testCreate() throws Exception{ //创建永久节点 client.create().forPath("/curator","/curator data".getBytes()); //创建永久有序节点 client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath("/curator_sequential","/curator_sequential data".getBytes()); //创建临时节点 client.create().withMode(CreateMode.EPHEMERAL) .forPath("/curator/ephemeral","/curator/ephemeral data".getBytes()); //创建临时有序节点 client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL) .forPath("/curator/ephemeral_path1","/curator/ephemeral_path1 data".getBytes()); client.create().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL) .forPath("/curator/ephemeral_path2","/curator/ephemeral_path2 data".getBytes()); } }

           在Curator中,首先需要通过CuratorFrameworkFactory创建连接实例,接着才能进行各种基本操作。需要说明的是,关于连接重试策略,Curator默认提供了以下几种:

    ExponentialBackoffRetry:重试一定次数,每次重试时间依次递增RetryNTimes:重试N次RetryOneTime:重试一次RetryUntilElapsed:重试一定时间

    1、所谓永久节点:ZooKeeper中生成的节点名跟我们在代码中设置的节点名是一样的,而有序节点跟普通节点不同之处在于ZooKeeper会自动在生成的节点名后面添加序列号,具体效果如下:

    2、临时节点:顾名思义ZooKeeper会在节点生成一段时间之后自动删除该节点

           此外,在创建临时有序节点的时候可以使用withProtection() 方法,该方法的作用是在创建的节点名前面添加GUID标识,其目的是为了避免出现“节点创建成功,但是ZooKeeper服务器在创建的节点名被返回给client前就出现了异常,从而导致临时节点没有被立即删除,而client也没法判断哪些节点被创建成功(注:创建的临时节点名有一定随机性)”的情况。具体效果如下:

    2、检查某个节点是否存在

    在上面代码中添加以下单元测试

    /** * 测试检查某个节点是否存在 * @throws Exception */ @Test public void testCheck() throws Exception{ Stat stat1 = client.checkExists().forPath("/curator"); Stat stat2 = client.checkExists().forPath("/curator2"); System.out.println("'/curator'是否存在: " + (stat1 != null ? true : false)); System.out.println("'/curator2'是否存在: " + (stat2 != null ? true : false)); }

    输出如下:

    '/curator'是否存在: true '/curator2'是否存在: false

    3、获取和设置节点数据

    继续添加以下单元测试:

    /** * 测试获取和设置节点数据 * @throws Exception */ @Test public void testGetAndSet() throws Exception{ //获取某个节点的所有子节点 System.out.println(client.getChildren().forPath("/")); //获取某个节点数据 System.out.println(new String(client.getData().forPath("/curator"))); //设置某个节点数据 client.setData().forPath("/curator","/curator modified data".getBytes()); }

    输出如下:

    [curator, curator_sequential0000000022, website, test, zookeeper, book] /curator data

    4、异步设置节点数据以及获取通知

    继续添加以下单元测试:

    /** * 测试异步设置节点数据 * @throws Exception */ @Test public void testSetDataAsync() throws Exception{ //创建监听器 CuratorListener listener = new CuratorListener() { @Override public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception { System.out.println(event.getPath()); } }; //添加监听器 client.getCuratorListenable().addListener(listener); //异步设置某个节点数据 client.setData().inBackground().forPath("/curator","/curator modified data with Async".getBytes()); //为了防止单元测试结束从而看不到异步执行结果,因此暂停10秒 Thread.sleep(10000); }

    当然,除了上述方式之外还有另一种异步执行获取通知的方式:

    /** * 测试另一种异步执行获取通知的方式 * @throws Exception */ @Test public void testSetDataAsyncWithCallback() throws Exception{ BackgroundCallback callback = new BackgroundCallback() { @Override public void processResult(CuratorFramework client, CuratorEvent event) throws Exception { System.out.println(event.getPath()); } }; //异步设置某个节点数据 client.setData().inBackground(callback).forPath("/curator","/curator modified data with Callback".getBytes()); //为了防止单元测试结束从而看不到异步执行结果,因此暂停10秒 Thread.sleep(10000); }

    5、删除节点

    继续添加以下单元测试:

    /** * 测试删除节点 * @throws Exception */ @Test public void testDelete() throws Exception{ //创建测试节点 client.create().orSetData().creatingParentContainersIfNeeded() .forPath("/curator/del_key1","/curator/del_key1 data".getBytes()); client.create().orSetData().creatingParentContainersIfNeeded() .forPath("/curator/del_key2","/curator/del_key2 data".getBytes()); client.create().forPath("/curator/del_key2/test_key","test_key data".getBytes()); //删除该节点 client.delete().forPath("/curator/del_key1"); //级联删除子节点 client.delete().guaranteed().deletingChildrenIfNeeded().forPath("/curator/del_key2"); }

    针对上面的单元测试,简单说明一下:

    orSetData()方法:如果节点存在则Curator将会使用给出的数据设置这个节点的值,相当于 setData() 方法creatingParentContainersIfNeeded()方法:如果指定节点的父节点不存在,则Curator将会自动级联创建父节点guaranteed()方法:如果服务端可能删除成功,但是client没有接收到删除成功的提示,Curator将会在后台持续尝试删除该节点deletingChildrenIfNeeded()方法:如果待删除节点存在子节点,则Curator将会级联删除该节点的子节点

    6、事务管理

    继续添加以下单元测试:

    /** * 测试事务管理:碰到异常,事务会回滚 * @throws Exception */ @Test public void testTransaction() throws Exception{ //定义几个基本操作 CuratorOp createOp = client.transactionOp().create() .forPath("/curator/one_path","some data".getBytes()); CuratorOp setDataOp = client.transactionOp().setData() .forPath("/curator","other data".getBytes()); CuratorOp deleteOp = client.transactionOp().delete() .forPath("/curator"); //事务执行结果 List<CuratorTransactionResult> results = client.transaction() .forOperations(createOp,setDataOp,deleteOp); //遍历输出结果 for(CuratorTransactionResult result : results){ System.out.println("执行结果是: " + result.getForPath() + "--" + result.getType()); } }

    在上面的测试代码中,很显然因为节点“/curator”存在子节点,所以在删除的时候将会报错,以上代码并不能成功执行。这时观察ZK节点,可以发现事务已经回滚了:

    7、命名空间

    继续添加以下单元测试:

    /** * 测试命名空间 * @throws Exception */ @Test public void testNamespace() throws Exception{ //创建带命名空间的连接实例 CuratorFramework client2 = CuratorFrameworkFactory.builder() .namespace("mydemo/v1") .connectString(SERVER) .sessionTimeoutMs(SESSION_TIMEOUT) .connectionTimeoutMs(CONNECTION_TIMEOUT) .retryPolicy(retryPolicy) .build(); //启动 client2.start(); client2.create().orSetData().creatingParentContainersIfNeeded() .forPath("/server1/method1","some data".getBytes()); client2.close(); } @After public void close(){ client.close(); }

    为了避免多个应用的节点名称冲突的情况,CuratorFramework提供了命名空间的概念。具体做法是:CuratorFramework会为它的API调用的节点路径的前面自动添加上命名空间。命名空间本质上是从根节点开始的一个路径,如:mydemo、mydemo/v1

    运行上面代码之后,效果如下:

    Processed: 0.011, SQL: 9