Java实现ZooKeeper的zNode监控
Java实现ZooKeeper的zNode监控
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"?>
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
4.0.0
tech.codestory.research
zookeeper
1.0.0-SNAPSHOT
<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>
3 log4j2.xml
在项目的 src/main/resources 下创建一个文件 log4j2.xml,内容为
<?xml version="1.0" encoding="UTF-8"?>
<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>
<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>
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
*/
@Slf4j
public 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:null
10: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:/watcher
10:15:43:463 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 00, version=0
10:15:50:906 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
10:15:50:910 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 01, version=1
10:15:56:004 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
10:15:56:007 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 02, version=2
10:16:00:246 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
10:16:00:249 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 03, version=3
10:16:06:399 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
10:16:06:402 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 10, version=4
10:16:10:217 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
10:16:10:220 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 11, version=5
10:16:14:444 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
10:16:14:447 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 12, version=6
10:16:20:118 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDeleted path:/watcher
10:16:20:118 [INFO] - ZooKeeperWatcher.run(101) - /watcher 被删除,退出
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
js设计模式之单例模式
js设计模式之单例模式1、定义保证一个类仅生成一个实例,并可以全局访问。 2、应用范围单列模式应用非常广泛,有些场景下某些对象只需要一个,比如浏览器中的window对象,全局缓存对象等。在实际开发过程中应用也比较多,比如点击一个按钮产生一个登陆框,无论点击多少次都应该只在第一次产生这个登录框,而在后面点击N次也只能是复用这个登陆框,而非重新创建,这样的场景就可以使用单列模式来创建这个登陆框。 3、实现一个简单的单例模式//简单版单列模式const Singleton = function (name) { this.name = name //这个instance来保存生成的实例 this.instance = null }Singleton.prototype.getName = function () { return this.name } //静态方法Singleton.getInstance = function (name) { if (!this.instance){ this.instance = new Singleton(name) } return this.in...
- 下一篇
python调用支付宝支付接口
<input type="text" name="price"> <input type="submit">
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
-
Docker使用Oracle官方镜像安装(12C,18C,19C)
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS8编译安装MySQL8.0.19
- Docker快速安装Oracle11G,搭建oracle11g学习环境
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- MySQL8.0.19开启GTID主从同步CentOS8
- CentOS7,8上快速安装Gitea,搭建Git服务器
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
推荐阅读
最新文章
- SpringBoot2整合Redis,开启缓存,提高访问速度
- Windows10,CentOS7,CentOS8安装Nodejs环境
- CentOS7编译安装Cmake3.16.3,解决mysql等软件编译问题
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
- SpringBoot2更换Tomcat为Jetty,小型站点的福音
- CentOS7设置SWAP分区,小内存服务器的救世主
- Docker安装Oracle12C,快速搭建Oracle学习环境
- Hadoop3单机部署,实现最简伪集群