经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Zookeeper » 查看文章
Java实现ZooKeeper的zNode监控
来源：jb51　　时间：2019/8/15 13:43:36　　对本文有异议

本文章默认你已经完成了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的内容

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.  <modelVersion>4.0.0</modelVersion>
  5.  <groupId>tech.codestory.research</groupId>
  6.  <artifactId>zookeeper</artifactId>
  7.  <version>1.0.0-SNAPSHOT</version>
  8.  <dependencies>
  9.   <dependency>
  10.    <groupId>org.apache.zookeeper</groupId>
  11.    <artifactId>zookeeper</artifactId>
  12.    <version>3.5.5</version>
  13.    <exclusions>
  14.     <exclusion>
  15.      <groupId>log4j</groupId>
  16.      <artifactId>log4j</artifactId>
  17.     </exclusion>
  18.     <exclusion>
  19.      <groupId>org.slf4j</groupId>
  20.      <artifactId>slf4j-log4j12</artifactId>
  21.     </exclusion>
  22.    </exclusions>
  23.   </dependency>
  24.   <dependency>
  25.    <groupId>org.apache.logging.log4j</groupId>
  26.    <artifactId>log4j-core</artifactId>
  27.    <version>2.12.1</version>
  28.   </dependency>
  29.   <dependency>
  30.    <groupId>org.apache.logging.log4j</groupId>
  31.    <artifactId>log4j-api</artifactId>
  32.    <version>2.12.1</version>
  33.   </dependency>
  34.   <dependency>
  35.    <groupId>org.apache.logging.log4j</groupId>
  36.    <artifactId>log4j-web</artifactId>
  37.    <version>2.12.1</version>
  38.   </dependency>
  39.   <dependency>
  40.    <groupId>org.apache.logging.log4j</groupId>
  41.    <artifactId>log4j-slf4j-impl</artifactId>
  42.    <version>2.12.1</version>
  43.   </dependency>
  44.   <dependency>
  45.    <groupId>org.slf4j</groupId>
  46.    <artifactId>slf4j-api</artifactId>
  47.    <version>1.7.28</version>
  48.   </dependency>
  49.   <dependency>
  50.    <groupId>org.slf4j</groupId>
  51.    <artifactId>slf4j-ext</artifactId>
  52.    <version>1.7.28</version>
  53.   </dependency>
  54.   <dependency>
  55.    <groupId>org.projectlombok</groupId>
  56.    <artifactId>lombok</artifactId>
  57.    <version>1.18.8</version>
  58.    <scope>provided</scope>
  59.   </dependency>
  60.  </dependencies>
  61. </project>

3   log4j2.xml

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

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <configuration status="DEBUG" name="codeStoryLogger">
  3.  <appenders>
  4.   <Console name="stdout">
  5.    <ThresholdFilter level="TRACE" onMatch="ACCEPT" onMismatch="DENY"/>
  6.    <PatternLayout pattern="%d{HH:mm:ss:SSS} [%p] - %c{1}.%M(%L) - %m%n"/>
  7.   </Console>
  8.  </appenders>
  9.  <ThresholdFilter level="TRACE"/>
  10.  <loggers>
  11.   <logger name="org.apache.zookeeper.ClientCnxn" level="ERROR" additivity="false">
  12.    <appender-ref ref="stdout" />
  13.   </logger>
  14.   <logger name="org.apache.zookeeper" level="TRACE" additivity="false">
  15.    <appender-ref ref="stdout"/>
  16.   </logger>
  17.   <logger name="tech.codestory" level="TRACE" additivity="false">
  18.    <appender-ref ref="stdout"/>
  19.   </logger>
  20.   <root level="WARN">
  21.    <appender-ref ref="stdout"/>
  22.   </root>
  23.  </loggers>
  24. </configuration>

4   创建ZooKeeper连接

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

ZooKeeper构造函数的定义

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

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

  1. public class ZooKeeperWatcher implements Watcher {
  2.  /** ZooKeeper的客户端连接 */
  3.  ZooKeeper zk;
  4.  public ZooKeeperWatcher(String hostPort, String zNode) throws KeeperException, IOException {
  5.   zk = new ZooKeeper(hostPort, 3000, this);
  6.   try {
  7.    Stat exists = zk.exists(zNode, true);
  8.    if(exists == null){
  9.     log.info(“{} 不存在”, zNode)
  10.    }
  11.   } catch (InterruptedException e) {
  12.    log.error("InterruptedException", e);
  13.   }
  14.  }
  15. }

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

  1. java.net.SocketException: Socket is not connected
  2.  at sun.nio.ch.Net.translateToSocketException(Net.java:162) ~[?:?]
  3.  at sun.nio.ch.Net.translateException(Net.java:196) ~[?:?]
  4.  at sun.nio.ch.Net.translateException(Net.java:202) ~[?:?]
  5.  at sun.nio.ch.SocketAdaptor.shutdownInput(SocketAdaptor.java:400) ~[?:?]
  6.  at org.apache.zookeeper.ClientCnxnSocketNIO.cleanup(ClientCnxnSocketNIO.java:198) [zookeeper-3.5.5.jar:3.5.5]
  7.  at org.apache.zookeeper.ClientCnxn$SendThread.cleanup(ClientCnxn.java:1338) [zookeeper-3.5.5.jar:3.5.5]
  8.  at org.apache.zookeeper.ClientCnxn$SendThread.cleanAndNotifyState(ClientCnxn.java:1276) [zookeeper-3.5.5.jar:3.5.5]
  9.  at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1254) [zookeeper-3.5.5.jar:3.5.5]
  10. Caused by: java.nio.channels.NotYetConnectedException
  11.  at sun.nio.ch.SocketChannelImpl.shutdownInput(SocketChannelImpl.java:917) ~[?:?]
  12.  at sun.nio.ch.SocketAdaptor.shutdownInput(SocketAdaptor.java:398) ~[?:?]
  13.  ... 4 more
  14. NotYetConnectedException的字面意思是连接还没有创建好，网络搜索了一下，建立连接需要一些时间，创建zk对象后马上调用exists命令，这时候连接还没有创建好，就会抛异常。ZooKeeper在连接建立成功之后，会发送一个WatchedEvent事件，我们可以利用这个事件完成建立连接的过程。修改后的代码如下，顺便添加了slf4j-ext中的Profiler，用于记录所消耗的时间。
  15. public class ZooKeeperWatcher implements Watcher {
  16.  /** 等待连接建立成功的信号 */
  17.  private CountDownLatch connectedSemaphore = new CountDownLatch(1);
  18.  /** ZooKeeper的客户端连接 */
  19.  ZooKeeper zk;
  20.  public ZooKeeperWatcher(String hostPort, String zNode) throws KeeperException, IOException {
  21.   Profiler profiler = new Profiler("连接到ZooKeeper");
  22.   profiler.start("开始链接");
  23.   zk = new ZooKeeper(hostPort, 3000, this);
  24.   try {
  25.      profiler.start("等待连接成功的Event");
  26.    connectedSemaphore.await();
  27.    Stat exists = zk.exists(zNode, true);
  28.    if(exists == null){
  29.     log.info(“{} 不存在”, zNode)
  30.    }
  31.   } catch (InterruptedException e) {
  32.    log.error("InterruptedException", e);
  33.   }
  34.   profiler.stop();
  35.   profiler.setLogger(log);
  36.   profiler.log();
  37.  }
  38.  /** 收到ZooKeeper的WatchedEvent */
  39.  @Override
  40.  public void process(WatchedEvent event) {
  41.   log.info("event = {}", event);
  42.   if (Event.EventType.None.equals(event.getType())) {
  43.    // 连接状态发生变化
  44.    if (Event.KeeperState.SyncConnected.equals(event.getState())) {
  45.     // 连接建立成功
  46.     connectedSemaphore.countDown();
  47.    }
  48.   }
  49.  }
  50. }

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

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

5   读取WatchedEvent

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


  1. if (Event.EventType.NodeDataChanged.equals(event.getType())
  2.   || Event.EventType.NodeCreated.equals(event.getType())) {
  3.  String path = event.getPath();
  4.  if (path != null && path.equals(zNode)) {
  5.   // 节点数据被修改
  6.   readNodeData();
  7.  }
  8. }
  9. /** 读节点数据 */
  10. private void readNodeData() {
  11.  try {
  12.   Stat stat = new Stat();
  13.   byte[] data = zk.getData(zNode, true, stat);
  14.   if (data != null) {
  15.    log.info("{}, value={}, version={}", zNode, new String(data), stat.getVersion());
  16.   }
  17.  } catch (KeeperException e) {
  18.   log.info("{} 不存在", zNode);  
  19.  } catch (InterruptedException e) {
  20.   log.error("InterruptedException", e);
  21.  }
  22. }

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

6   调整后的完整程序清单

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

  1. package tech.codestory.zookeeper.watcher;
  2. import java.io.IOException;
  3. import java.util.concurrent.CountDownLatch;
  4. import org.apache.zookeeper.*;
  5. import org.apache.zookeeper.data.Stat;
  6. import org.slf4j.profiler.Profiler;
  7. import lombok.extern.slf4j.Slf4j;
  8. /**
  9.  * 用于测试 ZooKeeper的 WatchedEvent用法
  10.  * @author code story
  11.  * @date 2019/8/13
  12.  */
  13. @Slf4j
  14. public class ZooKeeperWatcher implements Watcher, Runnable {
  15.  /** 等待连接建立成功的信号 */
  16.  private CountDownLatch connectedSemaphore = new CountDownLatch(1);
  17.  /** 退出系统的信号 */
  18.  static Integer quitSemaphore = Integer.valueOf(-1);
  19.  String zNode;
  20.  ZooKeeper zk;
  21.  public ZooKeeperWatcher(String hostPort, String zNode) throws KeeperException, IOException {
  22.   this.zNode = zNode;
  23.   Profiler profiler = new Profiler("连接到ZooKeeper");
  24.   profiler.start("开始链接");
  25.   zk = new ZooKeeper(hostPort, 3000, this);
  26.   try {
  27.    profiler.start("等待连接成功的Event");
  28.    connectedSemaphore.await();
  29.   } catch (InterruptedException e) {
  30.    log.error("InterruptedException", e);
  31.   }
  32.   profiler.stop();
  33.   profiler.setLogger(log);
  34.   profiler.log();
  35.   // 先读当前的数据
  36.   readNodeData();
  37.  }
  38.  /** 收到ZooKeeper的WatchedEvent */
  39.  @Override
  40.  public void process(WatchedEvent event) {
  41.   log.info("event = {}", event);
  42.   if (Event.EventType.None.equals(event.getType())) {
  43.    // 连接状态发生变化
  44.    if (Event.KeeperState.SyncConnected.equals(event.getState())) {
  45.     // 连接建立成功
  46.     connectedSemaphore.countDown();
  47.    }
  48.   } else if (Event.EventType.NodeDataChanged.equals(event.getType())
  49.     || Event.EventType.NodeCreated.equals(event.getType())) {
  50.    String path = event.getPath();
  51.    if (path != null && path.equals(zNode)) {
  52.     // 节点数据被修改
  53.     readNodeData();
  54.    }
  55.   } else if (Event.EventType.NodeDeleted.equals(event.getType())) {
  56.    String path = event.getPath();
  57.    if (path != null && path.equals(zNode)) {
  58.     synchronized (quitSemaphore) {
  59.      // 节点被删除，通知退出线程
  60.      quitSemaphore.notify();
  61.     }
  62.    }
  63.   }
  64.  }
  65.  /** 读节点数据 */
  66.  private void readNodeData() {
  67.   try {
  68.    Stat stat = new Stat();
  69.    byte[] data = zk.getData(zNode, true, stat);
  70.    if (data != null) {
  71.     log.info("{}, value={}, version={}", zNode, new String(data), stat.getVersion());
  72.    }
  73.   } catch (KeeperException e) {
  74.    log.info("{} 不存在", zNode);
  75.    try {
  76.     // 目的是添加Watcher
  77.     zk.exists(zNode, true);
  78.    } catch (KeeperException ex) {
  79.    } catch (InterruptedException ex) {
  80.    }
  81.   } catch (InterruptedException e) {
  82.    log.error("InterruptedException", e);
  83.   }
  84.  }
  85.  @Override
  86.  public void run() {
  87.   try {
  88.    synchronized (quitSemaphore) {
  89.     quitSemaphore.wait();
  90.     log.info("{} 被删除，退出", zNode);
  91.    }
  92.   } catch (InterruptedException e) {
  93.    log.error("InterruptedException", e);
  94.   }
  95.  }
  96.  /** 主程序，代码中写死了Server地址和zNode名，也可以改成从args中读取 */
  97.  public static void main(String[] args) {
  98.   String hostPort = "192.168.5.128:2181";
  99.   String zNode = "/watcher";
  100.   try {
  101.    new ZooKeeperWatcher(hostPort, zNode).run();
  102.   } catch (Exception e) {
  103.    log.error("new ZooKeeperExecutor()", e);
  104.   }
  105.  }
  106. }

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

  1. 10:13:31:979 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:None path:null
  2. 10:13:31:982 [DEBUG] - ZooKeeperWatcher.log(201) -
  3. + Profiler [连接到ZooKeeper]
  4. |-- elapsed time          [开始链接]  210.193 milliseconds.
  5. |-- elapsed time      [等待连接成功的Event] 9385.467 milliseconds.
  6. |-- Total         [连接到ZooKeeper] 9596.196 milliseconds.
  7. 10:13:31:996 [INFO] - ZooKeeperWatcher.readNodeData(84) - /watcher 不存在
  8. 10:15:43:451 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeCreated path:/watcher
  9. 10:15:43:463 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 00, version=0
  10. 10:15:50:906 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
  11. 10:15:50:910 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 01, version=1
  12. 10:15:56:004 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
  13. 10:15:56:007 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 02, version=2
  14. 10:16:00:246 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
  15. 10:16:00:249 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 03, version=3
  16. 10:16:06:399 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
  17. 10:16:06:402 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 10, version=4
  18. 10:16:10:217 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
  19. 10:16:10:220 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 11, version=5
  20. 10:16:14:444 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDataChanged path:/watcher
  21. 10:16:14:447 [INFO] - ZooKeeperWatcher.readNodeData(81) - /watcher, value=hello zk 12, version=6
  22. 10:16:20:118 [INFO] - ZooKeeperWatcher.process(50) - event = WatchedEvent state:SyncConnected type:NodeDeleted path:/watcher
  23. 10:16:20:118 [INFO] - ZooKeeperWatcher.run(101) - /watcher 被删除，退出

总结

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

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Zookeeper热门文章

  • java Zookeeper简述
  • zookeeper-操作与应用场景-《每日五分钟搞定大数据》
  • ZooInspector 连接不到 Zookeeper 的解决方法
  • IDEA zookeeper插件的使用
  • 解决dubbo错误ip及ip乱入问题的方法
  • zookeeper集群搭建
  • 大数据开发phoenix连接hbase流程详解
  • 快来看看 Dubbo 是如何实现服务的动态发现的
  • zookeeper特性与节点说明
  • zookeeper【1】配置管理

    • Zookeeper推荐文章

  • Zookeeper安装部署
  • 得物 ZooKeeper SLA 也可以 99.99%
  • zookeeper源码(06)ZooKeeperServer及子类
  • Dubbo 我手写几行代码，就把通信模式给你解释清楚！
  • Rpc-实现Zookeeper注册中心
  • Dubbo 中 Zookeeper 注册中心原理分析
  • Java&nbsp;zookeeper服务的使用详解
  • 【入门】ZooKeeper 相关概念总结
  • zookeeper特性与节点说明
  • 快来看看 Dubbo 是如何实现服务的动态发现的

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号