专栏首页程序猿讲故事Java实现ZooKeeper的zNode监控

Java实现ZooKeeper的zNode监控

上一篇文章已经完成了ZooKeeper的基本搭建和使用的介绍,现在开始用代码说话。参考 https://zookeeper.apache.org/doc/current/javaExample.html ,但对场景和代码都做了简化,只实现基本的Watcher功能。

1 场景设计

目的是体验ZooKeeper的Watcher功能。程序监控ZooKeeper的/watcher节点数据变化,当创建或修改数据时,控制台打印当前的数据内容和版本号;当/watcher被删除时,程序退出。

/watcher的创建、修改和删除操作,使用控制台或zkui操作。

2 搭建Maven项目

代码相对比较简单,就不用SpringBoot这个大杀器了,使用一个普通的Maven项目即可。

ZooKeeper客户端使用官方提供的Java库,org.apache.zookeeper: zookeeper: 3.5.5。日志框架使用习惯的slf4j+log4j2,ZooKeeper缺省使用的是log4j V1,因此在引入的时候需要excludes。另外,使用了lombok来简化一些代码。

以下是pom.xml的内容

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>tech.codestory.research</groupId> <artifactId>zookeeper</artifactId> <version>1.0.0-SNAPSHOT</version> <dependencies> <dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.5.5</version> <exclusions> <exclusion> <groupId>log4j</groupId> <artifactId>log4j</artifactId> </exclusion> <exclusion> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>2.12.1</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-api</artifactId> <version>2.12.1</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-web</artifactId> <version>2.12.1</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-slf4j-impl</artifactId> <version>2.12.1</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.28</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-ext</artifactId> <version>1.7.28</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.8</version> <scope>provided</scope> </dependency> </dependencies></project>

3 log4j2.xml

在项目的 src/main/resources 下创建一个文件 log4j2.xml,内容为

<?xml version="1.0" encoding="UTF-8"?><configuration status="DEBUG" name="codeStoryLogger"> <appenders> <Console name="stdout"> <ThresholdFilter level="TRACE" onMatch="ACCEPT" onMismatch="DENY"/> <PatternLayout pattern="%d{HH:mm:ss:SSS} [%p] - %c{1}.%M(%L) - %m%n"/> </Console> </appenders> <ThresholdFilter level="TRACE"/> <loggers> <logger name="org.apache.zookeeper.ClientCnxn" level="ERROR" additivity="false"> <appender-ref ref="stdout" /> </logger> <logger name="org.apache.zookeeper" level="TRACE" additivity="false"> <appender-ref ref="stdout"/> </logger> <logger name="tech.codestory" level="TRACE" additivity="false"> <appender-ref ref="stdout"/> </logger> <root level="WARN"> <appender-ref ref="stdout"/> </root> </loggers></configuration>

4 创建ZooKeeper连接

创建连接代码比较简单,只需要创建 ZooKeeper对象就行,

ZooKeeper构造函数的定义

/** * 创建一个 ZooKeeper 客户端对象 * @param connectString 逗号分隔的 host:port 字符串, * 单点如 127.0.0.1:2181, * 集群如 192.168.5.128:2181,192.168.5.129:2181,192.168.5.130:2181, * 还可以指定根节点,如 127.0.0.1:2181/foo/bar * @param sessionTimeout 毫秒为单位的超时时间 * @param watcher Watcher对象,用于接收 MatcherEvent * @throws IOException 网络错误时抛出异常 * @throws IllegalArgumentException 如果Root路径设置错误 */public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException;

写一段测试代码,创建zk对象后判断某一个zNode是否存在。

public class ZooKeeperWatcher implements Watcher { /** ZooKeeper的客户端连接 */ ZooKeeper zk; public ZooKeeperWatcher(String hostPort, String zNode) throws KeeperException, IOException { zk = new ZooKeeper(hostPort, 3000, this); try { Stat exists = zk.exists(zNode, true); if(exists == null){ log.info(“{} 不存在”, zNode) } } catch (InterruptedException e) { log.error("InterruptedException", e); } }}

运行这段代码,发现会抛异常

java.net.SocketException: Socket is not connected at sun.nio.ch.Net.translateToSocketException(Net.java:162) ~[?:?] at sun.nio.ch.Net.translateException(Net.java:196) ~[?:?] at sun.nio.ch.Net.translateException(Net.java:202) ~[?:?] at sun.nio.ch.SocketAdaptor.shutdownInput(SocketAdaptor.java:400) ~[?:?] at org.apache.zookeeper.ClientCnxnSocketNIO.cleanup(ClientCnxnSocketNIO.java:198) [zookeeper-3.5.5.jar:3.5.5] at org.apache.zookeeper.ClientCnxn$SendThread.cleanup(ClientCnxn.java:1338) [zookeeper-3.5.5.jar:3.5.5] at org.apache.zookeeper.ClientCnxn$SendThread.cleanAndNotifyState(ClientCnxn.java:1276) [zookeeper-3.5.5.jar:3.5.5] at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1254) [zookeeper-3.5.5.jar:3.5.5]Caused by: java.nio.channels.NotYetConnectedException at sun.nio.ch.SocketChannelImpl.shutdownInput(SocketChannelImpl.java:917) ~[?:?] at sun.nio.ch.SocketAdaptor.shutdownInput(SocketAdaptor.java:398) ~[?:?] ... 4 more

NotYetConnectedException的字面意思是连接还没有创建好,网络搜索了一下,建立连接需要一些时间,创建zk对象后马上调用exists命令,这时候连接还没有创建好,就会抛异常。ZooKeeper在连接建立成功之后,会发送一个WatchedEvent事件,我们可以利用这个事件完成建立连接的过程。修改后的代码如下,顺便添加了slf4j-ext中的Profiler,用于记录所消耗的时间。

public class ZooKeeperWatcher implements Watcher { /** 等待连接建立成功的信号 */ private CountDownLatch connectedSemaphore = new CountDownLatch(1); /** ZooKeeper的客户端连接 */ ZooKeeper zk; public ZooKeeperWatcher(String hostPort, String zNode) throws KeeperException, IOException { Profiler profiler = new Profiler("连接到ZooKeeper"); profiler.start("开始链接"); zk = new ZooKeeper(hostPort, 3000, this); try { profiler.start("等待连接成功的Event"); connectedSemaphore.await(); Stat exists = zk.exists(zNode, true); if(exists == null){ log.info(“{} 不存在”, zNode) } } catch (InterruptedException e) { log.error("InterruptedException", e); } profiler.stop(); profiler.setLogger(log); profiler.log(); } /** 收到ZooKeeper的WatchedEvent */ @Override public void process(WatchedEvent event) { log.info("event = {}", event); if (Event.EventType.None.equals(event.getType())) { // 连接状态发生变化 if (Event.KeeperState.SyncConnected.equals(event.getState())) { // 连接建立成功 connectedSemaphore.countDown(); } } }}

修改代码之后的执行记录日志如下,可以看到等待连接成功的Event耗时9秒多。网络上有文章说关闭防火墙可以秒连,但我测试过,没发现有什么变化,使用systemctl stop firewalld之后重新执行程序,仍然需要9秒多。

[INFO] - ZooKeeperWatcher.process(61) - event = WatchedEvent state:SyncConnected type:None path:null[DEBUG] - ZooKeeperWatcher.log(201) -+ Profiler [连接到ZooKeeper]|-- elapsed time [开始链接] 78.912 milliseconds.|-- elapsed time [等待连接成功的Event] 9330.606 milliseconds.|-- Total [连接到ZooKeeper] 9409.926 milliseconds. [INFO] - ZooKeeperWatcher.readNodeData(95) - /watcher 不存在

5 读取WatchedEvent

前面的代码,只是处理了建立连接成功时的Event,下面再来看看读取数据的过程。关键代码如下:

if (Event.EventType.NodeDataChanged.equals(event.getType()) || Event.EventType.NodeCreated.equals(event.getType())) { String path = event.getPath(); if (path != null && path.equals(zNode)) { // 节点数据被修改 readNodeData(); }} /** 读节点数据 */private void readNodeData() { try { Stat stat = new Stat(); byte[] data = zk.getData(zNode, true, stat); if (data != null) { log.info("{}, value={}, version={}", zNode, new String(data), stat.getVersion()); } } catch (KeeperException e) { log.info("{} 不存在", zNode); } catch (InterruptedException e) { log.error("InterruptedException", e); }}

当接收到创建节点和修改节点的WatchedEvent,都会将数据读出并打印在控制台。

6 调整后的完整程序清单

对前面的代码做了部分调整,同时添加了退出系统的机制:节点被删除。

package tech.codestory.zookeeper.watcher; import java.io.IOException;import java.util.concurrent.CountDownLatch;import org.apache.zookeeper.*;import org.apache.zookeeper.data.Stat;import org.slf4j.profiler.Profiler;import lombok.extern.slf4j.Slf4j; /** * 用于测试 ZooKeeper的 WatchedEvent用法 * @author code story * @date 2019/8/13 */@Slf4jpublic class ZooKeeperWatcher implements Watcher, Runnable { /** 等待连接建立成功的信号 */ private CountDownLatch connectedSemaphore = new CountDownLatch(1); /** 退出系统的信号 */ static Integer quitSemaphore = Integer.valueOf(-1); String zNode; ZooKeeper zk; public ZooKeeperWatcher(String hostPort, String zNode) throws KeeperException, IOException { this.zNode = zNode; Profiler profiler = new Profiler("连接到ZooKeeper"); profiler.start("开始链接"); zk = new ZooKeeper(hostPort, 3000, this); try { profiler.start("等待连接成功的Event"); connectedSemaphore.await(); } catch (InterruptedException e) { log.error("InterruptedException", e); } profiler.stop(); profiler.setLogger(log); profiler.log(); // 先读当前的数据 readNodeData(); } /** 收到ZooKeeper的WatchedEvent */ @Override public void process(WatchedEvent event) { log.info("event = {}", event); if (Event.EventType.None.equals(event.getType())) { // 连接状态发生变化 if (Event.KeeperState.SyncConnected.equals(event.getState())) { // 连接建立成功 connectedSemaphore.countDown(); } } else if (Event.EventType.NodeDataChanged.equals(event.getType()) || Event.EventType.NodeCreated.equals(event.getType())) { String path = event.getPath(); if (path != null && path.equals(zNode)) { // 节点数据被修改 readNodeData(); } } else if (Event.EventType.NodeDeleted.equals(event.getType())) { String path = event.getPath(); if (path != null && path.equals(zNode)) { synchronized (quitSemaphore) { // 节点被删除,通知退出线程 quitSemaphore.notify(); } } } } /** 读节点数据 */ private void readNodeData() { try { Stat stat = new Stat(); byte[] data = zk.getData(zNode, true, stat); if (data != null) { log.info("{}, value={}, version={}", zNode, new String(data), stat.getVersion()); } } catch (KeeperException e) { log.info("{} 不存在", zNode); try { // 目的是添加Watcher zk.exists(zNode, true); } catch (KeeperException ex) { } catch (InterruptedException ex) { } } catch (InterruptedException e) { log.error("InterruptedException", e); } } @Override public void run() { try { synchronized (quitSemaphore) { quitSemaphore.wait(); log.info("{} 被删除,退出", zNode); } } catch (InterruptedException e) { log.error("InterruptedException", e); } } /** 主程序,代码中写死了Server地址和zNode名,也可以改成从args中读取 */ public static void main(String[] args) { String hostPort = "192.168.5.128:2181"; String zNode = "/watcher"; try { new ZooKeeperWatcher(hostPort, zNode).run(); } catch (Exception e) { log.error("new ZooKeeperExecutor()", e); } }}

做一个测试,应用启动后创建节点,修改多次节点,最后删除节点,日志输出如下:

10:13:31:979 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:None path:null10:13:31:982 [DEBUG] - ZooKeeperWatcher.log(201) -+ Profiler [连接到ZooKeeper]|-- elapsed time [开始链接] 210.193 milliseconds.|-- elapsed time [等待连接成功的Event] 9385.467 milliseconds.|-- Total [连接到ZooKeeper] 9596.196 milliseconds. 10:13:31:996 [INFO] - ZooKeeperWatcher.readNodeData(84) - /watcher 不存在10:15:43:451 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeCreated path:/watcher10:15:43:463 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 00, version=010:15:50:906 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher10:15:50:910 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 01, version=110:15:56:004 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher10:15:56:007 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 02, version=210:16:00:246 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher10:16:00:249 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 03, version=310:16:06:399 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher10:16:06:402 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 10, version=410:16:10:217 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher10:16:10:220 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 11, version=510:16:14:444 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher10:16:14:447 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 12, version=610:16:20:118 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDeleted path:/watcher10:16:20:118 [INFO] - ZooKeeperWatcher.run(101) - /watcher 被删除,退出

本文分享自微信公众号 - 程序猿讲故事(codestory),作者:程序猿讲故事

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2019-08-15

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 安装配置ZooKeeper及基本用法

    要想学习分布式应用,ZooKeeper是一个绕不过去的基础系统。它为大型分布式计算提供开源的分布式配置服务、同步服务和命名注册。

    程序猿讲故事
  • MAT工具在MacBook的安装

    当Java应用出现内存溢出的问题的时候,需要拿工具分析dump文件的。JDK自带的jvisualvm和jhat都可以使用,另外还有一个工具是 Memory An...

    程序猿讲故事
  • ZooKeeper实现同步屏障(Barrier)

    按照维基百科的解释:同步屏障(Barrier)是并行计算中的一种同步方法。对于一群进程或线程,程序中的一个同步屏障意味着任何线程/进程执行到此后必须等待,直到所...

    程序猿讲故事
  • 聊聊nifi的BinlogEventListener

    nifi-1.11.4/nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-proce...

    codecraft
  • python 基础 —— 获取当前时间的

    py3study
  • 有趣的 CSS 像素艺术

    原文:fun-times-css-pixel-art 译者:nzbin 友情提示:由于国内网络的原因,CodePen可能会打不开或者非常慢,请耐心等待! ...

    叙帝利
  • ft6236 触摸屏驱动

    在目录下amp\a53_linux\drv\extdrv\touchpad\ft6236下可以看到ft6236.c的文件

    233333
  • 响铃:互联网金融理财端寡头时代来临,到底谁该被剩下来?

      近日,互联网金融又遇大事,百度与中信银行成立直销银行:百信银行,这意味着BAT三大巨头均以不同形式拿到银行牌照。至此互联网金融“井喷式”发展过后,进入了新的...

    曾响铃
  • 响铃:互联网金融理财端寡头时代来临,到底谁该被剩下来?

      近日,互联网金融又遇大事,百度与中信银行成立直销银行:百信银行,这意味着BAT三大巨头均以不同形式拿到银行牌照。至此互联网金融“井喷式”发展过后,进入了新的...

    曾响铃
  • 大数据开发:HDFS的故障恢复和高可用

    作为分布式文件系统的HDFS,在Hadoop技术生态当中,始终是不容忽视的。HDFS的稳定性和可靠性,对于后续的数据处理环节,提供底层支持,起着至关重要的作用。...

    成都加米谷大数据

扫码关注云+社区

领取腾讯云代金券