|
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);
}
}
}
|