Java实现ZooKeeper的zNode监控

2024-12-18 03:19:05

上一篇文章已经完成了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
 */
@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 被删除，退出

总结

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

使用JMX监控Zookeeper状态Java API

一.背景上一篇通过Java自带的JConsole来获取zookeeper状态.主要有几个不方便的地方,zk集群一般会部署3或者5台,在多个JConsole窗口中切换比较麻烦,各个zk服务及历史数据之间,不能直观比较.一般会做一个WEB管理页面来展示集群状态,设置报警阀值来做报警. 二.JVM平台提供Mbeans 在Java5.0以上版本,有一组API可以让Java应用程序和允许的工具监视和管理Java虚拟机(JVM)和虚拟机所在的本机操作系统.该组API在 java.lang.manageme
浅谈Java(SpringBoot)基于zookeeper的分布式锁实现

通过zookeeper实现分布式锁 1.创建zookeeper的client 首先通过CuratorFrameworkFactory创建一个连接zookeeper的连接CuratorFramework client public class CuratorFactoryBean implements FactoryBean<CuratorFramework>, InitializingBean, DisposableBean { private static final Logger LOGG
windows下zookeeper配置java环境变量的方法

先找到文件 zookeeper的bin目录下编辑zkEnv.cmd 修改如下图如果你的系统有多个JAVA_HOME类型的系统变量则需要在该文件中指定用哪一个,并且还要修改windows的jdk环境变量总结以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,谢谢大家对我们的支持.如果你想了解更多相关内容请查看下面相关链接
java使用zookeeper实现的分布式锁示例

使用zookeeper实现的分布式锁分布式锁,实现了Lock接口复制代码代码如下: package com.concurrent; import java.io.IOException;import java.util.ArrayList;import java.util.Collections;import java.util.List;import java.util.concurrent.CountDownLatch;import java.util.concurrent.TimeU
ZooKeeper Java API编程实例分析

本实例我们用的是java3.4.6版本,实例方便大家学习完后有不明白的可以在留言区讨论. 开发应用程序的ZooKeeper Java绑定主要由两个Java包组成: org.apache.zookeeper org.apache.zookeeper.data org.apache.zookeeper包由ZooKeeper监视的接口定义和ZooKeeper的各种回调处理程序组成. 它定义了ZooKeeper客户端类库的主要类以及许多ZooKeeper事件类型和状态的静态定义. org.apache.
java 中 zookeeper简单使用

一.zookeeper的基本原理数据模型,如下: ZooKeeper数据模型的结构与Unix文件系统很类似,整体上可以看作是一棵树,每个节点称做一个ZNode.每个ZNode都可以通过其路径唯一标识,比如上图中第三层的第一个ZNode,它的路径是/app1/c1.在每个ZNode上可存储少量数据(默认是1M, 可以通过配置修改,通常不建议在ZNode上存储大量的数据),这个特性非常有用.另外,每个ZNode上还存储了其Acl信息,这里需要注意,虽说ZNode的树形结构跟Unix文件系统很类似,
Java实现ZooKeeper的zNode监控

上一篇文章已经完成了ZooKeeper的基本搭建和使用的介绍,现在开始用代码说话.参考 https://zookeeper.apache.org/doc/current/javaExample.html ,但对场景和代码都做了简化,只实现基本的Watcher功能. 1 场景设计目的是体验ZooKeeper的Watcher功能.程序监控ZooKeeper的/watcher节点数据变化,当创建或修改数据时,控制台打印当前的数据内容和版本号:当/watcher被删除时,程序退出. /watche
Java之Zookeeper注册中心原理剖析

RPC框架中有3个重要的角色: 注册中心 :保存所有服务的名字,服务提供者的IP列表,服务消费者的IP列表服务提供者: 提供跨进程服务服务消费者: 寻找到指定命名的服务并消费. Zookeeper用作注册中心简单来讲,zookeeper可以充当一个服务注册表(Service Registry),让多个服务提供者形成一个集群,让服务消费者通过服务注册表获取具体的服务访问地址(IP+端口)去访问具体的服务提供者.如下图所示: 具体来说,zookeeper就是个分布式文件系统,每当一个服务提供者
java实现人工智能化屏幕监控窗口

本文实例为大家分享了java实现人工智能化屏幕监控窗口的具体代码,供大家参考,具体内容如下具体代码实现(含注释) public class Main{ public static void main(String[] args) throws Exception{ /* test code */ } /** *用于实时监控屏幕的窗口 *@author chengxi *@param void *@return void */ public static void mvcontroll() th
如何使用Java操作Zookeeper

简介 Java操作Zookeeper有很多种方式,如zookeeper.zkclient.curator等等,下面介绍下使用zkclient的方式操作Zookeeper. Maven依赖: <dependency> <groupId>com.101tec</groupId> <artifactId>zkclient</artifactId> <version>0.11</version> </dependency&g
Java调用Zookeeper的实现步骤

目录 watch机制常用API JAVA调用 watch机制 Zookeeper watch是一种监听通知机制,可以随时监听一些数据的变化,从而实现数据的及时性. Zookeeper所有的读操作getData(), getChildren()和 exists()都可以设置监听(watch).[写操作则是不能设置监视点的.] Watch的三个关键点: 一次有效:当设置监视的数据发生改变时,该监视事件会被发送到客户端,并且该监听将会停止,除非重启注册监听: 顺序保证:网络延迟或者其他因素可能导致不
java连接zookeeper的3种方式小结

目录 java连接zookeeper3种方式 1.使用zookeeper原始api 2.使用ZkClient客户端连接,这种连接比较简单 3.使用curator连接 Java集成zookeeper笔记一.引入zookeeper-3.4.5.jar(原生zk包) 二.手写ZookeeperBase.java java连接zookeeper3种方式 1.使用zookeeper原始api public class Demo { private static String ip = "192.168.
java连接zookeeper实现zookeeper教程

目录 java连接zookeeper实现zookeeper Java服务端连接Zookeeper,进行节点信息的获取,管理-整理成一个基本工具添加依赖: <dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.3.6</version> </dependency> 具体
解决java连接zookeeper很慢的问题

目录 java连接zookeeper很慢记一次惨痛的zookeeper连接教训 java连接zookeeper很慢最近在学习zookeeper,但是在连接zookeeper服务端时很慢,每次都超过18s 在头条尝试搜索了一下这个问题,找到了答案原因是在服务器端集群时,设置的host是使用hostname,而不是ip,且客户端没有配置hostname和ip的映射导致的问题, 在客户端加上第一张图中的ip映射关系即可,或者把服务端的server中的hostname换成ip即可记一次惨痛的zo
java.nio.file.WatchService 实时监控文件变化的示例代码

目录 1.示例代码 2.其实并没有实时在平时的开发过程中,会有很多场景需要实时监听文件的变化,如下:1.通过实时监控 mysql 的 binlog 日志实现数据同步2.修改配置文件后,希望系统可以实时感知3.应用系统将日志写入文件中,日志监控系统可以实时抓取日志,分析日志内容并进行报警4.类似 ide 工具,可以实时感知管理的工程下的文件变更在 Java 语言中,从 JDK7 开始,新增了java.nio.file.WatchService类,用来实时监控文件的变化. 1.示例代码 File

Java实现ZooKeeper的zNode监控

相关推荐

随机推荐