ZooKeeper 实战(四) Curator Watch事件监听

2024-01-13 18:36
文章标签 实战 监听 事件 zookeeper watch curator

本文主要是介绍ZooKeeper 实战(四) Curator Watch事件监听，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

  • ZooKeeper 实战(四) Curator Watch事件监听
    • 0.前言
    • 1.Watch 事件监听概念
    • 2.NodeCache
      • 2.1.全参构造器参数
      • 2.2.代码DEMO
      • 2.3.日志输出
    • 3.PathChildrenCache
      • 3.1.全参构造器参数
      • 3.2.子节点监听时间类型
      • 3.2.代码DEMO
    • 4.TreeCache
      • 4.1.构造器参数
      • 4.2.代码DEMO
      • 4.3.日志输出

ZooKeeper 实战(四) Curator Watch事件监听

0.前言

上一篇博客只介绍了有关Curator中对ZNode的CRUD操作，从本篇起开始逐步介绍更加高级的API操作。

1.Watch 事件监听概念

ZooKeeper 中引入了Watcher机制来实现了发布/订阅功能能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。虽然ZooKeeper原生支持通过注册Watcher来进行事件监听，但是其使用并不是特别方便，需要开发人员反复注册Watcher，比较繁琐。

而 Curator 引入了Cache 来实现对 ZooKeeper 服务端事件的监听。

Curator 中提供了三种 Cache（Watcher）来监听不同节点变化类型：

  • NodeCache：监听指定的节点。
  • PathChildrenCache：监听指定节点的子节点。
  • TreeCache：监听指定节点及其子孙节点。

2.NodeCache

监听指定的节点，增删改都会监听。

2.1.全参构造器参数

/*** @param: client 注册监听的客户端* @param: path 节点路径* @param: dataIsCompressed 是否开启数据压缩。传递的数据会进行压缩,传递速度快,但取数据时需要把压缩的数据进行转换,默认为false*/
public NodeCache(CuratorFramework client, String path, boolean dataIsCompressed);

2.2.代码DEMO

    @Overridepublic void run(ApplicationArguments args) throws Exception {log.info("。。。。。。。。。。。。。容器初始化完毕。。。。。。。。。。。。。。");String path = "/ahao/watcher";TimeUnit.SECONDS.sleep(3);// 创建NodeCache对象NodeCache nodeCache = new NodeCache(client,path);// 添加监听器nodeCache.getListenable().addListener(new NodeCacheListener() {@Overridepublic void nodeChanged() throws Exception {ChildData currentData = nodeCache.getCurrentData();if (currentData != null){String s = new String(currentData.getData(),StandardCharsets.UTF_8);log.info("监听{}节点发生变化，数据内容：{}",path,s);}else {log.info("监听{}节点被删除了",path);}}});// 开启监听nodeCache.start();TimeUnit.SECONDS.sleep(2);// 创建节点client.create().creatingParentsIfNeeded().forPath(path,"第一次新增".getBytes(StandardCharsets.UTF_8));TimeUnit.SECONDS.sleep(2);// 更新节点client.setData().forPath(path,"数据修改了".getBytes(StandardCharsets.UTF_8));TimeUnit.SECONDS.sleep(2);// 删除节点client.delete().deletingChildrenIfNeeded().forPath(path);}

2.3.日志输出

在这里插入图片描述

3.PathChildrenCache

监听指定节点的子节点。当一个子节点增删改时， PathChildrenCache会包含最新的子节点的数据和状态。

3.1.全参构造器参数

/*** @param: client 注册监听的客户端* @param: path 节点路径* @param: cacheData 是否缓存节点内容（包含节点状态）* @param: dataIsCompressed 是否开启数据压缩。传递的数据会进行压缩,传递速度快,但取数据时需要把压缩的数据进行转换,默认为false* @param: executorService 用于PathChildrenCache的后台线程的线程池。该线程池应该是单线程的，否则缓存可能会看到不一致的结果*/
public PathChildrenCache(CuratorFramework client, String path, boolean cacheData, boolean dataIsCompressed, final CloseableExecutorService executorService)

3.2.子节点监听时间类型

public enum Type
{// 子节点添加CHILD_ADDED,// 子节点的数据变更CHILD_UPDATED,// 子节点被删除CHILD_REMOVED,// 以下三个事件类型表示：当连接断开时，PathChildrenCache将继续保持其断开连接之前的状态，并且在连接恢复后，PathChildrenCache将为断开连接期间发生的所有添加、删除和更新发出正常的子事件。// 当连接状态处于ConnectionState.SUSPENDED。CONNECTION_SUSPENDED,// 当连接状态处于ConnectionState.RECONNECTEDCONNECTION_RECONNECTED,// 当连接状态处于ConnectionState.LOSTCONNECTION_LOST,// 当通过PathChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT)启动监听时，该事件表示PathChildrenCache初始化完成This event signals that the initial cache has been populated.INITIALIZED
}

3.2.代码DEMO

    @Overridepublic void run(ApplicationArguments args) throws Exception {log.info("。。。。。。。。。。。。。容器初始化完毕。。。。。。。。。。。。。。");String path = "/ahao/watcher";TimeUnit.SECONDS.sleep(3);// 创建PathChildrenCache对象// 此处的cacheData参数一定要设置为true，不然Curator不会缓存数据当本地，// 那么后续pathChildrenCache.getCurrentData()得到的数据都为nullPathChildrenCache pathChildrenCache = new PathChildrenCache(client,path,true);// 添加监听器pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {@Overridepublic void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {if (event.getType() == PathChildrenCacheEvent.Type.INITIALIZED){log.info("PathChildrenCache初始化完，事件类型：{}", event.getType());}else {ChildData currentData = event.getData();log.info("事件类型：{}，监听到的子节点发生变化：{}",event.getType(),currentData.getPath());}}});// 开启监听pathChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);// 创建子节点TimeUnit.SECONDS.sleep(2);client.create().creatingParentsIfNeeded().forPath(path+"/c1");client.create().creatingParentsIfNeeded().forPath(path+"/c2");client.create().creatingParentsIfNeeded().forPath(path+"/c3/age");// 修改子节点TimeUnit.SECONDS.sleep(2);client.setData().forPath(path+"/c1","c1更新了".getBytes(StandardCharsets.UTF_8));client.setData().forPath(path+"/c2","c2更新了".getBytes(StandardCharsets.UTF_8));// 删除子节点TimeUnit.SECONDS.sleep(2);client.delete().deletingChildrenIfNeeded().forPath(path+"/c3");}

3.3.日志输出

可以看出，PathChildrenCache只会监听直属子节点的变化，其非直属子节点的后代节点如/c3/age，没有发布通知。

在这里插入图片描述

4.TreeCache

监听指定节点及其子孙节点。

4.1.构造器参数

/*** @param: client 注册监听的客户端* @param: path 节点路径*/
public TreeCache(CuratorFramework client, String path)/*** @param: client 注册监听的客户端* @param: path 节点路径* @param: cacheData 是否缓存节点内容（包含节点状态）* @param: dataIsCompressed 是否开启数据压缩。传递的数据会进行压缩,传递速度快,但取数据时需要把压缩的数据进行转换,默认为false* @param: maxDepth 最大深度。最深的那个后代节点到path所需要经过的节点数* @param: executorService 用于PathChildrenCache的后台线程的线程池。该线程池应该是单线程的，否则缓存可能会看到不一致的结果* @param: createParentNodes 是否需要创建父节点。如果父节点不存在泽创建父节点（容器节点）* @param: TreeCacheSelector TreeCache选择器。根据指定的策略和条件，选择适合的缓存树来创建和维护TreeCache*/
TreeCache(CuratorFramework client, String path, boolean cacheData, boolean dataIsCompressed, int maxDepth, final ExecutorService executorService, boolean createParentNodes, TreeCacheSelector selector)

4.2.代码DEMO

    @Overridepublic void run(ApplicationArguments args) throws Exception {log.info("。。。。。。。。。。。。。容器初始化完毕。。。。。。。。。。。。。。");String path = "/ahao/watcher/tree";TimeUnit.SECONDS.sleep(3);// 创建TreeCache对象，也可通过TreeCache.newBuilder()创建TreeCache treeCache = new TreeCache(client,path);treeCache.getListenable().addListener(new TreeCacheListener() {@Overridepublic void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {if (event.getType() == TreeCacheEvent.Type.INITIALIZED){log.info("TreeCache初始化完，事件类型：{}", event.getType());}else {ChildData currentData = event.getData();log.info("事件类型：{}，监听到的子节点发生变化：{}",event.getType(),currentData.getPath());}}});// 开启监听treeCache.start();// 创建节点TimeUnit.SECONDS.sleep(2);client.create().creatingParentsIfNeeded().forPath(path);client.create().creatingParentsIfNeeded().forPath(path +"/t1");client.create().creatingParentsIfNeeded().forPath(path +"/t2/ccc");// 修改子节点TimeUnit.SECONDS.sleep(2);client.setData().forPath(path,"根节点更新了".getBytes(StandardCharsets.UTF_8));client.setData().forPath(path +"/t2/ccc","/t2/ccc更新了".getBytes(StandardCharsets.UTF_8));// 删除子节点TimeUnit.SECONDS.sleep(2);client.delete().deletingChildrenIfNeeded().forPath(path +"/t2");}

4.3.日志输出

可以看出TreeCache会监听当前节点和后代节点的变化。

在这里插入图片描述

这篇关于ZooKeeper 实战(四) Curator Watch事件监听的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/602407。

相关文章

Java 队列Queue从原理到实战指南

Java 队列Queue从原理到实战指南

《Java队列Queue从原理到实战指南》本文介绍了Java中队列（Queue）的底层实现、常见方法及其区别,通过LinkedList和ArrayDeque的实现,以及循环队列的概念,展示了如何高效... 目录一、队列的认识队列的底层与集合框架常见的队列方法插入元素方法对比（add和offer）移除元素方法
阅读更多...
Spring Boot基于 JWT 优化 Spring Security 无状态登录实战指南

Spring Boot基于 JWT 优化 Spring Security 无状态登录实战指南

《SpringBoot基于JWT优化SpringSecurity无状态登录实战指南》本文介绍如何使用JWT优化SpringSecurity实现无状态登录,提高接口安全性,并通过实际操作步骤... 目录Spring Boot 实战：基于 JWT 优化 Spring Security 无状态登录一、先搞懂：为什
阅读更多...
C++11中的包装器实战案例

C++11中的包装器实战案例

《C++11中的包装器实战案例》本文给大家介绍C++11中的包装器实战案例,本文结合实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录引言1.std::function1.1.什么是std::function1.2.核心用法1.2.1.包装普通函数1.2.
阅读更多...
Nginx概念、架构、配置与虚拟主机实战操作指南

Nginx概念、架构、配置与虚拟主机实战操作指南

《Nginx概念、架构、配置与虚拟主机实战操作指南》Nginx是一个高性能的HTTP服务器、反向代理服务器、负载均衡器和IMAP/POP3/SMTP代理服务器,它支持高并发连接,资源占用低,功能全面且... 目录Nginx 深度解析：概念、架构、配置与虚拟主机实战一、Nginx 的概念二、Nginx 的特点
阅读更多...
Spring IOC核心原理详解与运用实战教程

Spring IOC核心原理详解与运用实战教程

《SpringIOC核心原理详解与运用实战教程》本文详细解析了SpringIOC容器的核心原理,包括BeanFactory体系、依赖注入机制、循环依赖解决和三级缓存机制,同时,介绍了SpringBo... 目录1. Spring IOC核心原理深度解析1.1 BeanFactory体系与内部结构1.1.1
阅读更多...
Redis 命令详解与实战案例

Redis 命令详解与实战案例

《Redis命令详解与实战案例》本文详细介绍了Redis的基础知识、核心数据结构与命令、高级功能与命令、最佳实践与性能优化,以及实战应用场景,通过实战案例,展示了如何使用Redis构建高性能应用系统... 目录Redis 命令详解与实战案例一、Redis 基础介绍二、Redis 核心数据结构与命令1. 字符
阅读更多...
在SpringBoot+MyBatis项目中实现MySQL读写分离的实战指南

在SpringBoot+MyBatis项目中实现MySQL读写分离的实战指南

《在SpringBoot+MyBatis项目中实现MySQL读写分离的实战指南》在SpringBoot和MyBatis项目中实现MySQL读写分离,主要有两种思路：一种是在应用层通过代码和配置手动控制... 目录如何选择实现方案核心实现：应用层手动分离实施中的关键问题与解决方案总结在Spring Boot和
阅读更多...
Python AST 模块实战演示

Python AST 模块实战演示

《PythonAST模块实战演示》Python的ast模块提供了一种处理Python代码的强大工具,通过解析代码生成抽象语法树（AST）,可以进行代码分析、修改和生成,接下来通过本文给大家介绍Py... 目录 什么是抽象语法树（AST）️ ast 模块的核心用法1. 解析代码生成 AST2. 查看 AST
阅读更多...
C++ 多态性实战之何时使用 virtual 和 override的问题解析

C++ 多态性实战之何时使用 virtual 和 override的问题解析

《C++多态性实战之何时使用virtual和override的问题解析》在面向对象编程中,多态是一个核心概念,很多开发者在遇到override编译错误时,不清楚是否需要将基类函数声明为virt... 目录C++ 多态性实战：何时使用 virtual 和 override？引言问题场景判断是否需要多态的三个关
阅读更多...
MySQL 批量插入的原理和实战方法(快速提升大数据导入效率)

MySQL 批量插入的原理和实战方法(快速提升大数据导入效率)

《MySQL批量插入的原理和实战方法(快速提升大数据导入效率)》在日常开发中,我们经常需要将大量数据批量插入到MySQL数据库中,本文将介绍批量插入的原理、实现方法,并结合Python和PyMySQ... 目录一、批量插入的优势二、mysql 表的创建示例三、python 实现批量插入1. 安装 PyMyS
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2