当前位置： 首页 > news >正文

Zookeeper客户端工具 Apache Curator 最佳实践

news 2024/10/25 23:22:02

1、Apache Curator 概述

前面我们学习了Zookeeper相关的内容，今天我们就先来学习一个更好用的Zookeeper的客户端工具   Apache Curator。这个 客户端工具是在Zookeeper的JavaAPI的基础上做了封装，使得开发人员可以更高效的使用Zookeeper。老规矩，我们先来看看官方文档

官方文档： https://curator.apache.org/docs/about

打开官网我们就看到了一句很醒目的话  Guava is to Java What Curator is to Zookeeper 。可见这个客户端工具和Zookeeper的关系是很紧密的。

2、Apache Curator 使用

新建项目，加入以下依赖

     <!-- https://mvnrepository.com/artifact/org.apache.curator/curator-framework --><dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>5.7.0</version></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>5.7.0</version> <!-- 请检查最新版本 --></dependency>

需要注意的是我们Zookeeper的版本是3.9.2,这里我们使用5.7版本的curator是没问题的。

2.1、创建连接 

我们可以参照官方文档上的案例，来创建链接，代码如下

private String zookeeperConnectionString = "192.168.200.100:9002,192.168.200.100:9003,192.168.200.100:9004";public CuratorFramework getConnect(){RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);CuratorFramework client = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);client.start();return client;}

当然还有第二种方式创建链接

   private CuratorFramework client;public void getConnect(){RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);client = CuratorFrameworkFactory.builder().connectString(zookeeperConnectionString).sessionTimeoutMs(5000).retryPolicy(retryPolicy).build();client.start();}

通常为了更加直观，更推荐使用第二种方式去创建连接。 

2.2、数据基本操作

前面我们已经创建好了连接，后续我们就可以使用 这个连接操作zookeeper了。我们继续查看 文档上描述基本的CRUD的方法

根据上的描述，这里也给出相应的示例代码

/*** @Description 测试链接* @Author wcan* @Date 2024/10/25 上午 11:33* @ClassName ConnectTest* @Version 1.0*/
public class CuratorDemo {private String zookeeperConnectionString = "192.168.200.100:9002,192.168.200.100:9003,192.168.200.100:9004";private CuratorFramework client;public void getConnect(){RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);client = CuratorFrameworkFactory.builder().connectString(zookeeperConnectionString).sessionTimeoutMs(5000).retryPolicy(retryPolicy).build();//开启连接client.start();}/*** @Description 创建节点  并且指定节点上存放的数据* @Param [nodePath, value]* @return void* @Date 2024/10/24 下午 18:46* @Author wcan* @Version 1.0*/public void nodeCreate(String nodePath,String value) throws Exception {String path = client.create().forPath(nodePath, value.getBytes());System.out.println(path);}/*** @Description 如果创建节点，不指定数据，则默认保存客户端的ip* @Param [nodePath]* @return void* @Date 2024/10/24 下午 18:45* @Author wcan* @Version 1.0*/public void nodeCreate(String nodePath) throws Exception {String path = client.create().forPath(nodePath);System.out.println(path);}/*** @Description 创建多级节点 父节点不存在 则先创建父节点* @Param [nodePath]* @return void* @Date 2024/10/24 下午 18:45* @Author wcan* @Version 1.0*/public void createMultipleNode(String nodePath) throws Exception {String path = client.create().creatingParentsIfNeeded().forPath(nodePath);System.out.println(path);}/*** @Description 查询指定节点数据*   type =1 查询当前节点数据*   type=2 查询当前节点子节点数据*   type=3 查询当前节点状态* @Param [path, type]* @return void* @Date 2024/10/24 下午 18:49* @Author wcan* @Version 1.0*/public void queryNode(String path,int type) throws Exception {if(1==type){byte[] bytes = client.getData().forPath(path);System.out.println(new String(bytes));}if (2==type){List<String> strings = client.getChildren().forPath(path);strings.forEach(o-> System.out.println(o));}if (3==type){Stat stat = new Stat();byte[] bytes = client.getData().storingStatIn(stat).forPath(path);System.out.println(new String(bytes));System.out.println(stat);}}/*** @Description 给指定的节点赋值 指定字符集* @Param [path, data]* @return void* @Date 2024/10/24 下午 19:27* @Author wcan* @Version 1.0*/public void setNodeDate(String path,String data) throws Exception {byte[] bytes = data.getBytes("UTF-8");client.setData().forPath(path, bytes);}/*** @Description 删除节点* @Param [path]* @return void* @Date 2024/10/24 下午 19:37* @Author wcan* @Version 1.0*/public void deleteNodeDate(String path) throws Exception {//删除普通节点
//        client.delete().forPath(path);//删除带子节点的节点client.delete().deletingChildrenIfNeeded().forPath(path);}public static void main(String[] args) throws Exception {CuratorDemo connectTest = new CuratorDemo();connectTest.getConnect();
//        connectTest.nodeCreate("/jerry2","hi i‘m jerry ");
//        connectTest.nodeCreate("/tom");
//        connectTest.createMultipleNode("/jerry2/tom");
//        connectTest.queryNode("/jerry2",1);
//        connectTest.queryNode("/",2);
//        connectTest.queryNode("/jerry2",3);connectTest.setNodeDate("/jerry2","hello world");connectTest.queryNode("/jerry2",1);connectTest.deleteNodeDate("/jerry2");connectTest.client.close();}
}

上面的案例大家可以 自己尝试一下。

3、Watch机制

3.1、Curator的Cache

我们知道了基本的操作之后，就来看看 使用Curator中怎么编码使用Watch机制的，Curator引入了 Cache 来实现对 ZooKeeper 服务端事件的监听。所以我们可以查看官方文档上对Cache描述的章节

Curator中提供了四种 Cache  用来实现缓存某个节点的数据，客户端可以注册监听器，当数据发生变更的时候就会通知客户端。上图中有基本的介绍，大家可以自行查阅。这里我帮大家来梳理一下这四种缓存的区别，包括适用的场景

特性Curator CachePath CacheNode CacheTree Cache
监控范围灵活单个节点单个节点整个树结构
事件通知支持数据变化数据变化、节点存在/不存在节点增加/删除、数据变化
适用场景多种监控需求简单节点监控频繁访问单个节点复杂树形结构监控

3.2、Curator Cache

CuratorCache是一个接口，它可以监控单个节点也可以监控当前节点下的子节点，综合性能是最好的，也是最灵活的一种。具体的用法下面给出一段代码示例：

/*** @Description* @Author wcan* @Date 2024/10/24 下午 20:20* @ClassName CuratorCachesDemo* @Version 1.0*/
public class CuratorCachesDemo {private String zookeeperConnectionString = "192.168.200.100:9002,192.168.200.100:9003,192.168.200.100:9004";public static void main(String[] args) throws Exception {RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);CuratorFramework client = CuratorFrameworkFactory.builder().connectString(zookeeperConnectionString).sessionTimeoutMs(5000).retryPolicy(retryPolicy).build();//开启连接client.start();// 创建路径，如果不存在String path = "/tom";if (client.checkExists().forPath(path) == null) {client.create().creatingParentsIfNeeded().forPath(path);}// 创建 CuratorCacheCuratorCache cache = CuratorCache.build(client, path);// 添加事件监听器CuratorCacheListener listener = CuratorCacheListener.builder().forCreates(event -> System.out.println("创建节点：" + event.getPath())).forChanges((oldNode, newNode) -> System.out.println("修改节点：" + oldNode.getPath())).forDeletes(node -> System.out.println("删除节点：" + node.getPath())).build();// 注册监听器cache.listenable().addListener(listener);// 启动缓存cache.start();// 阻塞主线程Thread.sleep(Long.MAX_VALUE);// 最后关闭
//        cache.close();
//        client.close();}
}

我们运行上述代码，然后去修改zookeeper上的节点数据

然后观察控制台的输出信息

3.3、Path Cache

我们从这个描述信息里面可以知道，PathCache是被用来监控ZNode的。 每当添加、更新或删除子元素时，Path Cache都会改变其状态，以包含当前的子元素集、子元素的数据和子元素的状态。下面我们来上代码

/*** @Description* @Author wcan* @Date 2024/10/24 下午 23:05* @ClassName PathCacheDemo* @Version 1.0*/
public class PathCacheDemo {private static String zookeeperConnectionString = "XXXXXXX";public static void main(String[] args) throws Exception {RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);CuratorFramework client = CuratorFrameworkFactory.builder().connectString(zookeeperConnectionString).sessionTimeoutMs(5000).retryPolicy(retryPolicy).build();//开启连接client.start();// 创建路径，如果不存在String path = "/jerry";if (client.checkExists().forPath(path) == null) {client.create().creatingParentsIfNeeded().forPath(path);}PathChildrenCache cache = new PathChildrenCache(client,path,true);cache.getListenable().addListener((client1, event) -> {if (event.getType() == PathChildrenCacheEvent.Type.INITIALIZED) {System.out.println("Initialized cache for path: " + event.getData().getData());} else if (event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {System.out.println("Child added: " + event.getData().getPath());System.out.println("Data: " + new String(event.getData().getData()));} else if (event.getType() == PathChildrenCacheEvent.Type.CHILD_UPDATED) {System.out.println("Child updated: " + event.getData().getPath());System.out.println("Data: " + new String(event.getData().getData()));} else if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {System.out.println("Child removed: " + event.getData().getPath());}else {System.out.println("Unknown event type: " + event.getType());}});// 开始缓存cache.start();// 保持程序运行System.out.println("Press Enter to exit...");System.in.read();}
}

同样的我们也可以去zookeeper上 操作Jerry这个节点 然后观察控制台的输出

3.4、Node Cache

NodeCache用于监控一个单个节点，使用场景是高频访问的某个节点这里先给出代码

public class NodeCacheDemo {private static String zookeeperConnectionString = "XXXXXXX";public static void main(String[] args) throws Exception {RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);CuratorFramework client = CuratorFrameworkFactory.builder().connectString(zookeeperConnectionString).sessionTimeoutMs(5000).retryPolicy(retryPolicy).build();//开启连接client.start();//1. 创建NodeCache对象NodeCache nodeCache = new NodeCache(client,"/tom");//2. 注册监听nodeCache.getListenable().addListener(new NodeCacheListener() {@Overridepublic void nodeChanged() throws Exception {System.out.println("节点变化了~");//获取修改节点后的数据byte[] data = nodeCache.getCurrentData().getData();System.out.println(new String(data));}});//3. 开启监听.如果设置为true，则开启监听是，加载缓冲数据nodeCache.start(true);// 保持程序运行System.out.println("Press Enter to exit...");System.in.read();}
}

3.5、Tree Cache

TreeCache 是一种用于监控指定节点及其所有子节点的数据变化，适用于需要监控整个树形结构及其动态变化的情况。相关代码如下，大家可以自行测试

public class TreeCacheDemo {private static String zookeeperConnectionString = "XXXXXXX";public static void main(String[] args) throws Exception {RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);CuratorFramework client = CuratorFrameworkFactory.builder().connectString(zookeeperConnectionString).sessionTimeoutMs(5000).retryPolicy(retryPolicy).build();//开启连接client.start();TreeCache treeCache = new TreeCache(client,"/app2");//2. 注册监听treeCache.getListenable().addListener((client1, event) ->{System.out.println("节点变化了");System.out.println(event);System.out.println("节点路径：" + event.getData().getPath());System.out.println("节点数据：" + new String(event.getData().getData()));System.out.println("节点类型：" + event.getType());System.out.println("节点状态：" + event.getData().getStat());System.out.println("=======================================");});//3. 开启treeCache.start();// 保持程序运行System.out.println("Press Enter to exit...");System.in.read();}
}

4、实战案例

前面我们已经把Curator常用的一些api和Watch机制的实现都给大家介绍了，接着我们来做两个案例，巩固一下前面学习的内容

 4.1、分布式锁案例

我们来实现一个分布式锁的案例，设计思想还是和之前的那篇介绍Zookeeper的文章一样，这里我们使用Curator来实现。

加入依赖

    implementation 'org.apache.curator:curator-recipes:5.7.0'implementation 'org.apache.curator:curator-framework:5.7.0'

代码如下：

@RestController
public class UserInfoController {private final InterProcessMutex interProcessMutex;public UserInfoController(InterProcessMutex interProcessMutex) {this.interProcessMutex = interProcessMutex;}@RequestMapping("/getUserInfo")public Map getUserInfo() throws Exception {Map jerry = new HashMap<String, Object>();String msg = "";if (interProcessMutex.acquire(1000, TimeUnit.MILLISECONDS)) {msg = "当前线程： " + Thread.currentThread().getId() + " 获取到锁";try {jerry.put("name", Thread.currentThread().getName());jerry.put("age", 18);jerry.put("sex", "男");jerry.put("address", "深圳");} finally {try {interProcessMutex.release();} catch (Exception e) {throw new RuntimeException(e);}}} elsemsg = "当前线程： " + Thread.currentThread().getName() + " 没有获取到锁";jerry.put("msg", msg);return jerry;}}

为了方便这里我将分布式锁的配置直接加载启动类里

@SpringBootApplication
public class JerryStoreApplication {public static void main(String[] args) {SpringApplication.run(JerryStoreApplication.class, args);System.out.println("jerry-store started");}@Beanpublic CuratorFramework curatorFramework(){RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().connectString("ip:port").sessionTimeoutMs(5000).retryPolicy(retryPolicy).build();return curatorFramework;}@Beanpublic InterProcessMutex interProcessMutex(CuratorFramework curatorFramework){curatorFramework.start();return new InterProcessMutex(curatorFramework, "/tom");}}

启动程序就可以使用Jmeter工具进行测试了。

4.2、分布式配置中心实战

 我们的需求是将部分配置信息存放在Zookeeper上，当这些信息发生变更的时候，利用监听机制捕获到并通知到订阅的客户端，让客户端获取到最新的配置。实现逻辑如下

public class ConfigListener {private CuratorFramework client;private PathChildrenCache cache;public ConfigListener() throws Exception {client = CuratorFrameworkFactory.newClient(CommonConstant.ZK_URL, new ExponentialBackoffRetry(1000, 3));client.start();cache = new PathChildrenCache(client, "/config/appConfig", true);cache.start();cache.getListenable().addListener((client, event) -> {System.out.println("配置发生变更: " + event.getType());//此处可以将最新配置 重新赋值 //或者根据自己的业务需求执行一段特殊的逻辑 // TODObyte[] data = event.getData().getData();System.out.println(new String(data));});}public void close() throws Exception {cache.close();client.close();}public static void main(String[] args) throws Exception {ConfigListener listener = new ConfigListener();//阻塞Thread.sleep(Long.MAX_VALUE);}
}

/*** @Description* @Author wcan* @Date 2024/10/25 下午 17:50* @ClassName CommonConstant* @Version 1.0*/
public class CommonConstant {public static final String CONFIG_PATH = "/config/appConfig";public static final String ZK_URL = "192.168.200.100:9002,192.168.200.100:9003,192.168.200.100:9004";
}

我们最后再写一个客户端程序来测试

/*** @Description 分布式配置中心案例* @Author wcan* @Date 2024/10/25 下午 17:46* @ClassName ConfigCenter* @Version 1.0*/
public class ConfigCenter {private CuratorFramework client;public ConfigCenter() {client = CuratorFrameworkFactory.newClient(CommonConstant.ZK_URL, new ExponentialBackoffRetry(1000, 3));client.start();}public void setConfig(String key, String value) throws Exception {String path = CommonConstant.CONFIG_PATH + "/" + key;if (client.checkExists().forPath(path) == null) {client.create().creatingParentsIfNeeded().forPath(path, value.getBytes());} else {client.setData().forPath(path, value.getBytes());}}public String getConfig(String key) throws Exception {String path = CommonConstant.CONFIG_PATH + "/" + key;byte[] data = client.getData().forPath(path);return new String(data);}public void close() {client.close();}public static void main(String[] args) throws Exception {ConfigCenter configCenter = new ConfigCenter();// 设置配置configCenter.setConfig("redis.url", "192.168.200.100:6379");// 获取配置String dbUrl = configCenter.getConfig("redis.url");System.out.println("redis.url: " + dbUrl);configCenter.close();}
}

 启动这个测试类 ，观察ConfigListener程序的控制台输出，就能看到效果了，我们如果要实现自己的业务功能，就可以将自己需要的功能放在 TODO的部分。

5、总结

本篇文章给大家详细的介绍了Curator 这个Zookeeper的客户端工具，从API的使用，到Watch机制的实现方案，都给出了相应的代码，最后带着大家实现了一个分布式队列的案例，希望对大家有所帮助，有疑问的小伙伴欢迎留下自己的问题，或者在公众号留言均可。


http://www.mrgr.cn/news/58595.html

相关文章：

  • 【移动应用开发】界面设计（二）实现水果列表页面
  • C#从零开始学习(用户界面)(unity Lab4)
  • 【electron7】调试对话图片的加密处理
  • HTB：CozyHosting[WriteUP]
  • 高翔【自动驾驶与机器人中的SLAM技术】学习笔记（十二）拓展图优化库g2o（一）框架
  • 数据结构与算法——Java实现 47. 从中序与后序遍历序列构造二叉树
  • 10340 文本编辑器（vim）
  • Swift 是一种由苹果公司开发的强大而直观的编程语言，主要用于开发 iOS、macOS、watchOS 和 tvOS 等苹果平台的应用程序。
  • C++中如何使用文件系统路径
  • AcWing 89：a^b ← 快速幂
  • 136.只出现一次的数字
  • 【开源项目】经典开源项目数字孪生工地——开源工程及源码
  • fpga系列 HDL: 竞争和冒险 01
  • 计算机网络：网络层 —— IPv4 协议的表示方法及其编址方法
  • python 线程间通信用什么手段
  • 微软投资比特币：将总资产1%投资于BTC？股东投票决定最终结果！
  • 洛谷 P1060 [NOIP2006 普及组] 开心的金明
  • C++ 移动语义
  • Vue学习记录之二十 postcss自定义插件及Unocss的使用
  • 遇到这3种接口测试问题，其实，你可以这么办~
  • 混个1024勋章
  • 2023年12月中国电子学会青少年软件编程（图形化）等级考试试卷（二级）答案 + 解析
  • CMU生成式人工智能大模型：从入门到放弃（九）
  • CMU生成式人工智能大模型：从入门到放弃（八）
  • 电机的旋转原理和发电原理！
  • 永恒之蓝漏洞利用复现