Java代码操作Zookeeper（利用 Apache Curator 库）

1. Zookeeper原生客户端库存在的缺点

• 复杂性高：原生客户端库提供了底层的 API，需要开发者手动处理很多细节，如连接管理、会话管理、非常处理等。这增加了开发的复杂性，容易堕落。
  
• 连接管理繁琐：利用原生客户端库时，开发者需要手动管理与 ZooKeeper 的连接。这包罗连接的创建、重连、会话超时处理等。
  
• 非常处理复杂：原生客户端库的 API 抛出多种非常，如 KeeperException、InterruptedException 等。开发者需要手动处理这些非常，增加了代码的复杂性。
  
• 缺少高级功能：原生客户端库缺少一些高级功能，如连接池管理、自动重试、负载均衡等。这些功能在实际应用中非常有用，但需要开发者本身实现或利用第三方库。
  
• 缺少封装和抽象：原生客户端库提供了底层的 API，缺少更高层次的封装和抽象。开发者需要本身编写大量的代码来实现常见的功能，如分布式锁、配置管理等。
  
• 性能调优困难：原生客户端库的性能调优需要开发者手动进行，如调解连接超时时间、会话超时时间等。这需要对 ZooKeeper 的工作原理有深入的理解。
  
• 缺少社区支持：相比于一些更高级的客户端库（如 Curator），原生客户端库的社区支持相对较少。开发者在利用过程中遇到问题时，可能难以找到办理方案。
  

2. Apache Curator介绍

2.1 基本概述

• 界说：Apache Curator是专为Apache ZooKeeper设计的Java/JVM客户端库，通过提供高级API框架及一系列实用工具，大幅低落利用ZooKeeper的复杂度并提升应用的可靠性。
  
• 开发背景：Curator最初由Netflix公司开源，目前是Apache的顶级项目。
  

2.2 焦点功能

• 高可用性连接管理：自动处理与ZooKeeper服务器的连接断开和重新连接，确保连接的稳固性和可靠性。
  
• 易于利用的API：封装复杂的ZooKeeper原语，提供更直观、简洁的利用方式，低落开发难度。
  
• 模式（Recipes）：预置了一系列常见的分布式盘算模式，如leader选举、分布式锁、缓存机制等，开发者可以快速实现这些分布式体系经典难题。
  
• 服务发现与负载均衡：支持动态的服务注册与发现，便于构建云原生应用，进步体系的可扩展性和灵活性。
  
• 异步DSL：针对Java 8及以上版本提供了异步编程的支持，进步了相应速度和程序服从。
  

3. 利用指南

3.1 添加 Maven 依靠

1. <dependency>
2.     <groupId>org.apache.curator</groupId>
3.     <artifactId>curator-framework</artifactId>
4.     <version>2.12.0</version>
5. </dependency>
6. <dependency>
7.     <groupId>org.apache.curator</groupId>
8.     <artifactId>curator-recipes</artifactId>
9.     <version>2.12.0</version>
10. </dependency>

复制代码

3.2 创建 Curator 客户端

1. import org.apache.curator.framework.CuratorFramework;
2. import org.apache.curator.framework.CuratorFrameworkFactory;
3. import org.apache.curator.retry.ExponentialBackoffRetry;
4. import org.apache.zookeeper.CreateMode;
6. public class CuratorExample {
7.     public static void main(String[] args) throws Exception {
8.         String connectString = "192.168.200.138:2181";
9.         String path = "/curator1";
10.         byte[] data = "myData".getBytes();
12.         ExponentialBackoffRetry retry = new ExponentialBackoffRetry(5000, 10);
13.         // 创建 Curator 客户端
14.         CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retry);
16.         // 启动客户端
17.         client.start();
19.         // 创建节点
20.         client.create().withMode(CreateMode.PERSISTENT).forPath(path, data);
22.         // 获取节点数据
23.         byte[] retrievedData = client.getData().forPath(path);
24.         System.out.println("Retrieved data: " + new String(retrievedData));
26.         // 关闭客户端
27.         client.close();
28.     }
29. }

复制代码

3.3 增编削查操作及Watcher监听

1. import org.apache.curator.framework.CuratorFramework;
2. import org.apache.curator.framework.CuratorFrameworkFactory;
3. import org.apache.curator.framework.api.CuratorEvent;
4. import org.apache.curator.retry.ExponentialBackoffRetry;
5. import org.apache.zookeeper.CreateMode;
6. import org.apache.zookeeper.WatchedEvent;
7. import org.apache.zookeeper.Watcher;
9. public class CuratorExample {
10.     public static void main(String[] args) throws Exception {
11.         String connectString = "192.168.200.138:2181";
12.         String path = "/curator1";
13.         byte[] data1 = "myData1".getBytes();
14.         byte[] data2 = "myData2".getBytes();
16.         ExponentialBackoffRetry retry = new ExponentialBackoffRetry(5000, 10);
17.         // 创建 Curator 客户端
18.         CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retry);
20.         // 启动客户端
21.         client.start();
23.         // 增加临时监听操作
24.         client.getCuratorListenable().addListener((CuratorFramework c, CuratorEvent event) -> {
25.             switch (event.getType()) {
26.                 case WATCHED:
27.                     WatchedEvent watchedEvent = event.getWatchedEvent();
28.                     if (watchedEvent.getType() == Watcher.Event.EventType.NodeDataChanged) {
29.                         System.out.println("监听的数据变化为: " + new String(c.getData().forPath(path)));
30.                         System.out.println("触发事件");
31.                     }
32.             }
33.         });
35.         // 创建节点
36.         client.create().withMode(CreateMode.PERSISTENT).forPath(path, data1);
38.         // 获取节点数据
39.         byte[] retrievedData = client.getData().watched().forPath(path);
40.         System.out.println("原始数据: " + new String(retrievedData));
42.         // 修改节点数据
43.         client.setData().forPath(path, data2);
44.         Thread.sleep(2000);
46.         // 删除节点
47.         client.delete().forPath(path);
48.         Thread.sleep(2000);
49.     }
50. }

复制代码

3.4 进行永世监听

1. import org.apache.curator.framework.CuratorFramework;
2. import org.apache.curator.framework.CuratorFrameworkFactory;
3. import org.apache.curator.framework.recipes.cache.ChildData;
4. import org.apache.curator.framework.recipes.cache.NodeCache;
5. import org.apache.curator.retry.ExponentialBackoffRetry;
6. import org.apache.zookeeper.CreateMode;
8. public class PermanentWatcherExample {
9.     public static void main(String[] args) throws Exception {
10.         String connectString = "192.168.200.138:2181";
11.         String path = "/curator1";
12.         byte[] data1 = "myData1".getBytes();
13.         byte[] data2 = "myData2".getBytes();
14.         byte[] data3 = "myData3".getBytes();
16.         ExponentialBackoffRetry retry = new ExponentialBackoffRetry(5000, 10);
17.         // 创建 Curator 客户端
18.         CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retry);
20.         // 启动客户端
21.         client.start();
23.         // 永久监听
24.         client.create().withMode(CreateMode.PERSISTENT).forPath(path, data1);
25.         NodeCache nodeCache = new NodeCache(client, path);
26.         nodeCache.start();
27.         nodeCache.getListenable().addListener(() -> {
28.             ChildData currentData = nodeCache.getCurrentData();
29.             if (currentData != null) {
30.                 System.out.println("触发了永久监听的回调，当前值为：" + new String(currentData.getData()));
31.             }
32.         });
34.         client.setData().forPath(path, data1);
35.         Thread.sleep(2000);
37.         client.setData().forPath(path, data2);
38.         Thread.sleep(2000);
40.         client.setData().forPath(path, data3);
41.         Thread.sleep(2000);
43.         client.delete().forPath(path);
44.     }
45. }

复制代码

3.5 利用分布式锁

1. import org.apache.curator.framework.CuratorFramework;
2. import org.apache.curator.framework.CuratorFrameworkFactory;
3. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
4. import org.apache.curator.retry.ExponentialBackoffRetry;
6. public class DistributedLockExample {
7.     public static void main(String[] args) throws Exception {
8.         String connectString = "192.168.200.138:2181";
9.         String path = "/myLock";
11.         ExponentialBackoffRetry retry = new ExponentialBackoffRetry(5000, 10);
12.         // 创建 Curator 客户端
13.         CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retry);
15.         // 启动客户端
16.         client.start();
18.         // 创建分布式锁
19.         InterProcessMutex lock = new InterProcessMutex(client, path);
21.         // 获取锁
22.         lock.acquire();
23.         try {
24.             // 执行临界区代码
25.             System.out.println("Lock acquired, executing critical section...");
26.             Thread.sleep(2000);
27.         } finally {
28.             // 释放锁
29.             lock.release();
30.             System.out.println("Lock released.");
31.         }
33.         // 关闭客户端
34.         client.close();
35.     }
36. }

复制代码