Java实现ZooKeeper的zNode监控-低调大师

Java实现ZooKeeper的zNode监控

2019-08-14 602

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

原文地址https://www.cnblogs.com/codestory/p/11356426.html

微信关注我们

原文链接：https://yq.aliyun.com/articles/714434

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道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...

2019-08-15

656

2019-08-15

651

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。