- 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);
- }
- }
- }
来源: https://www.cnblogs.com/codestory/p/11356426.html