首页 > > 程序设计 > Java技术 >

Java实现ZooKeeper的zNode监控

2019-08-16 12:38:16来源：博客园 阅读 ()

新老客户大回馈,云服务器低至5折

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

 */

@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 被删除，退出

原文链接:https://www.cnblogs.com/codestory/p/11356426.html
如有疑问请与原作者联系

标签：

版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点，本站所提供的摄影照片，插画，设计作品，如需使用，请与原作者联系，版权归原作者所有

上一篇：SpringCloud高并发性能优化

下一篇：JavaWeb零基础入门-01 基础概念说明

相关文章

IDC资讯： 主机资讯 注册资讯 托管资讯 vps资讯 网站建设

网站运营： 建站经验 策划盈利 搜索优化 网站推广 免费资源

网站联盟： 联盟新闻 联盟介绍 联盟点评 网赚技巧

行业资讯： 搜索引擎 网络游戏 电子商务 广告传媒

网络编程： Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它

服务器技术： Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护

软件技巧： 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷

网页制作： FrontPages Dreamweaver Javascript css photoshop fireworks Flash

程序设计： Java技术 C/C++ VB delphi

网络知识： 网络协议 网络安全 网络管理 组网方案 Cisco技术

操作系统： Win2000 WinXP Win2003 Mac OS Linux FreeBSD

热门词条
最新资讯
热门关注
热门标签