Java实现ZooKeeper的zNode监控

2023-08-17 15:12:04 63

上一篇文章已经完成了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缺省使用的是log4jV1，因此在引入的时候需要excludes。另外，使用了lombok来简化一些代码。

以下是pom.xml的内容



4.0.0
tech.codestory.research
zookeeper
1.0.0-SNAPSHOT


org.apache.zookeeper
zookeeper
3.5.5


log4j
log4j


org.slf4j
slf4j-log4j12




org.apache.logging.log4j
log4j-core
2.12.1


org.apache.logging.log4j
log4j-api
2.12.1


org.apache.logging.log4j
log4j-web
2.12.1


org.apache.logging.log4j
log4j-slf4j-impl
2.12.1


org.slf4j
slf4j-api
1.7.28


org.slf4j
slf4j-ext
1.7.28


org.projectlombok
lombok
1.18.8
provided

3 log4j2.xml

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

4 创建ZooKeeper连接

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

ZooKeeper构造函数的定义

/**
*创建一个ZooKeeper客户端对象
*@paramconnectString逗号分隔的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
*@paramsessionTimeout毫秒为单位的超时时间
*@paramwatcherWatcher对象，用于接收MatcherEvent
*@throwsIOException网络错误时抛出异常
*@throwsIllegalArgumentException如果Root路径设置错误
*/
publicZooKeeper(StringconnectString,intsessionTimeout,Watcherwatcher)
throwsIOException;

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

publicclassZooKeeperWatcherimplementsWatcher{
/**ZooKeeper的客户端连接*/
ZooKeeperzk;
publicZooKeeperWatcher(StringhostPort,StringzNode)throwsKeeperException,IOException{
zk=newZooKeeper(hostPort,3000,this);
try{
Statexists=zk.exists(zNode,true);
if(exists==null){
log.info(“{}不存在”,zNode)
}
}catch(InterruptedExceptione){
log.error("InterruptedException",e);
}
}
}

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

java.net.SocketException:Socketisnotconnected
atsun.nio.ch.Net.translateToSocketException(Net.java:162)~[?:?]
atsun.nio.ch.Net.translateException(Net.java:196)~[?:?]
atsun.nio.ch.Net.translateException(Net.java:202)~[?:?]
atsun.nio.ch.SocketAdaptor.shutdownInput(SocketAdaptor.java:400)~[?:?]
atorg.apache.zookeeper.ClientCnxnSocketNIO.cleanup(ClientCnxnSocketNIO.java:198)[zookeeper-3.5.5.jar:3.5.5]
atorg.apache.zookeeper.ClientCnxn$SendThread.cleanup(ClientCnxn.java:1338)[zookeeper-3.5.5.jar:3.5.5]
atorg.apache.zookeeper.ClientCnxn$SendThread.cleanAndNotifyState(ClientCnxn.java:1276)[zookeeper-3.5.5.jar:3.5.5]
atorg.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1254)[zookeeper-3.5.5.jar:3.5.5]
Causedby:java.nio.channels.NotYetConnectedException
atsun.nio.ch.SocketChannelImpl.shutdownInput(SocketChannelImpl.java:917)~[?:?]
atsun.nio.ch.SocketAdaptor.shutdownInput(SocketAdaptor.java:398)~[?:?]
...4more
NotYetConnectedException的字面意思是连接还没有创建好，网络搜索了一下，建立连接需要一些时间，创建zk对象后马上调用exists命令，这时候连接还没有创建好，就会抛异常。ZooKeeper在连接建立成功之后，会发送一个WatchedEvent事件，我们可以利用这个事件完成建立连接的过程。修改后的代码如下，顺便添加了slf4j-ext中的Profiler，用于记录所消耗的时间。
publicclassZooKeeperWatcherimplementsWatcher{
/**等待连接建立成功的信号*/
privateCountDownLatchconnectedSemaphore=newCountDownLatch(1);
/**ZooKeeper的客户端连接*/
ZooKeeperzk;
publicZooKeeperWatcher(StringhostPort,StringzNode)throwsKeeperException,IOException{
Profilerprofiler=newProfiler("连接到ZooKeeper");
profiler.start("开始链接");
zk=newZooKeeper(hostPort,3000,this);
try{
profiler.start("等待连接成功的Event");
connectedSemaphore.await();
Statexists=zk.exists(zNode,true);
if(exists==null){
log.info(“{}不存在”,zNode)
}
}catch(InterruptedExceptione){
log.error("InterruptedException",e);
}
profiler.stop();
profiler.setLogger(log);
profiler.log();
}
/**收到ZooKeeper的WatchedEvent*/
@Override
publicvoidprocess(WatchedEventevent){
log.info("event={}",event);
if(Event.EventType.None.equals(event.getType())){
//连接状态发生变化
if(Event.KeeperState.SyncConnected.equals(event.getState())){
//连接建立成功
connectedSemaphore.countDown();
}
}
}
}

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

[INFO]-ZooKeeperWatcher.process(61)-event=WatchedEventstate:SyncConnectedtype:Nonepath:null
[DEBUG]-ZooKeeperWatcher.log(201)-
+Profiler[连接到ZooKeeper]
|--elapsedtime[开始链接]78.912milliseconds.
|--elapsedtime[等待连接成功的Event]9330.606milliseconds.
|--Total[连接到ZooKeeper]9409.926milliseconds.
[INFO]-ZooKeeperWatcher.readNodeData(95)-/watcher不存在

5 读取WatchedEvent

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

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

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

6 调整后的完整程序清单

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

packagetech.codestory.zookeeper.watcher;
importjava.io.IOException;
importjava.util.concurrent.CountDownLatch;
importorg.apache.zookeeper.*;
importorg.apache.zookeeper.data.Stat;
importorg.slf4j.profiler.Profiler;
importlombok.extern.slf4j.Slf4j;
/**
*用于测试ZooKeeper的WatchedEvent用法
*@authorcodestory
*@date2019/8/13
*/
@Slf4j
publicclassZooKeeperWatcherimplementsWatcher,Runnable{
/**等待连接建立成功的信号*/
privateCountDownLatchconnectedSemaphore=newCountDownLatch(1);
/**退出系统的信号*/
staticIntegerquitSemaphore=Integer.valueOf(-1);
StringzNode;
ZooKeeperzk;
publicZooKeeperWatcher(StringhostPort,StringzNode)throwsKeeperException,IOException{
this.zNode=zNode;
Profilerprofiler=newProfiler("连接到ZooKeeper");
profiler.start("开始链接");
zk=newZooKeeper(hostPort,3000,this);
try{
profiler.start("等待连接成功的Event");
connectedSemaphore.await();
}catch(InterruptedExceptione){
log.error("InterruptedException",e);
}
profiler.stop();
profiler.setLogger(log);
profiler.log();
//先读当前的数据
readNodeData();
}
/**收到ZooKeeper的WatchedEvent*/
@Override
publicvoidprocess(WatchedEventevent){
log.info("event={}",event);
if(Event.EventType.None.equals(event.getType())){
//连接状态发生变化
if(Event.KeeperState.SyncConnected.equals(event.getState())){
//连接建立成功
connectedSemaphore.countDown();
}
}elseif(Event.EventType.NodeDataChanged.equals(event.getType())
||Event.EventType.NodeCreated.equals(event.getType())){
Stringpath=event.getPath();
if(path!=null&&path.equals(zNode)){
//节点数据被修改
readNodeData();
}
}elseif(Event.EventType.NodeDeleted.equals(event.getType())){
Stringpath=event.getPath();
if(path!=null&&path.equals(zNode)){
synchronized(quitSemaphore){
//节点被删除，通知退出线程
quitSemaphore.notify();
}
}
}
}
/**读节点数据*/
privatevoidreadNodeData(){
try{
Statstat=newStat();
byte[]data=zk.getData(zNode,true,stat);
if(data!=null){
log.info("{},value={},version={}",zNode,newString(data),stat.getVersion());
}
}catch(KeeperExceptione){
log.info("{}不存在",zNode);
try{
//目的是添加Watcher
zk.exists(zNode,true);
}catch(KeeperExceptionex){
}catch(InterruptedExceptionex){
}
}catch(InterruptedExceptione){
log.error("InterruptedException",e);
}
}
@Override
publicvoidrun(){
try{
synchronized(quitSemaphore){
quitSemaphore.wait();
log.info("{}被删除，退出",zNode);
}
}catch(InterruptedExceptione){
log.error("InterruptedException",e);
}
}
/**主程序，代码中写死了Server地址和zNode名，也可以改成从args中读取*/
publicstaticvoidmain(String[]args){
StringhostPort="192.168.5.128:2181";
StringzNode="/watcher";
try{
newZooKeeperWatcher(hostPort,zNode).run();
}catch(Exceptione){
log.error("newZooKeeperExecutor()",e);
}
}
}

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

10:13:31:979[INFO]-ZooKeeperWatcher.process(50)-event=WatchedEventstate:SyncConnectedtype:Nonepath:null
10:13:31:982[DEBUG]-ZooKeeperWatcher.log(201)-
+Profiler[连接到ZooKeeper]
|--elapsedtime[开始链接]210.193milliseconds.
|--elapsedtime[等待连接成功的Event]9385.467milliseconds.
|--Total[连接到ZooKeeper]9596.196milliseconds.
10:13:31:996[INFO]-ZooKeeperWatcher.readNodeData(84)-/watcher不存在
10:15:43:451[INFO]-ZooKeeperWatcher.process(50)-event=WatchedEventstate:SyncConnectedtype:NodeCreatedpath:/watcher
10:15:43:463[INFO]-ZooKeeperWatcher.readNodeData(81)-/watcher,value=hellozk00,version=0
10:15:50:906[INFO]-ZooKeeperWatcher.process(50)-event=WatchedEventstate:SyncConnectedtype:NodeDataChangedpath:/watcher
10:15:50:910[INFO]-ZooKeeperWatcher.readNodeData(81)-/watcher,value=hellozk01,version=1
10:15:56:004[INFO]-ZooKeeperWatcher.process(50)-event=WatchedEventstate:SyncConnectedtype:NodeDataChangedpath:/watcher
10:15:56:007[INFO]-ZooKeeperWatcher.readNodeData(81)-/watcher,value=hellozk02,version=2
10:16:00:246[INFO]-ZooKeeperWatcher.process(50)-event=WatchedEventstate:SyncConnectedtype:NodeDataChangedpath:/watcher
10:16:00:249[INFO]-ZooKeeperWatcher.readNodeData(81)-/watcher,value=hellozk03,version=3
10:16:06:399[INFO]-ZooKeeperWatcher.process(50)-event=WatchedEventstate:SyncConnectedtype:NodeDataChangedpath:/watcher
10:16:06:402[INFO]-ZooKeeperWatcher.readNodeData(81)-/watcher,value=hellozk10,version=4
10:16:10:217[INFO]-ZooKeeperWatcher.process(50)-event=WatchedEventstate:SyncConnectedtype:NodeDataChangedpath:/watcher
10:16:10:220[INFO]-ZooKeeperWatcher.readNodeData(81)-/watcher,value=hellozk11,version=5
10:16:14:444[INFO]-ZooKeeperWatcher.process(50)-event=WatchedEventstate:SyncConnectedtype:NodeDataChangedpath:/watcher
10:16:14:447[INFO]-ZooKeeperWatcher.readNodeData(81)-/watcher,value=hellozk12,version=6
10:16:20:118[INFO]-ZooKeeperWatcher.process(50)-event=WatchedEventstate:SyncConnectedtype:NodeDeletedpath:/watcher
10:16:20:118[INFO]-ZooKeeperWatcher.run(101)-/watcher被删除，退出

总结

以上所述是小编给大家介绍的Java实现ZooKeeper的zNode监控,希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对毛票票网站的支持！
如果你觉得本文对你有帮助，欢迎转载，烦请注明出处，谢谢！

Java实现ZooKeeper的zNode监控

热门推荐

随机推荐