Zookeeper客户端工具 Apache Curator 最佳实践
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 Cache | Path Cache | Node Cache | Tree 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机制的实现方案,都给出了相应的代码,最后带着大家实现了一个分布式队列的案例,希望对大家有所帮助,有疑问的小伙伴欢迎留下自己的问题,或者在公众号留言均可。