ZooKeeper监听器原理
ZooKeeper 监听器（Watcher）是 ZooKeeper 提供的一种机制，用于实现分布式系统中的事件通知。
1.流程
- 注册监听器：
- 客户端在与 ZooKeeper 服务器建立连接后，可以通过某些操作（例如获取节点数据、检查节点是否存在等）注册一个 Watcher 对象，来监视特定节点的状态变化。
- 事件通知：
- 当某个与客户端关联的节点状态发生变化（例如节点被创建、数据被修改、节点被删除等），ZooKeeper 服务器会将相应的事件通知发送给客户端。
- 事件队列：
- ZooKeeper 客户端维护一个事件队列，用于存储接收到的事件通知。
- 当事件通知到达客户端时，会被放入事件队列中等待处理。
- 回调处理：
- 客户端在注册监听器时，通常会提供一个回调函数（或者说是回调方法）。
- 当事件通知被放入事件队列后，客户端会从事件队列中取出通知，并执行相应的回调函数来处理事件。
- 重新注册监听器：
- 一旦监听器收到了事件通知并处理完毕，客户端通常会选择重新注册监听器，以便继续监视节点的状态变化。这样就能保持持续的事件监听。
- 临时性：
- 监听器是一次性的，即一旦触发了事件通知，该监听器就会被移除。
- 因此，客户端需要在处理完事件后重新注册监听器，以确保持续地监视节点状态的变化。
通过这种机制，ZooKeeper 可以实现分布式系统中的事件驱动模型，使得客户端能够实时地获取和响应节点状态的变化，从而实现更加灵活和可靠的分布式协作。
2.示例
本示例创建了一个 ZooKeeperWatcherExample 类，其中使用了 ZooKeeper 的 Java 客户端 API 来连接 ZooKeeper 服务器、创建节点、设置节点数据，并注册了节点监听器。当节点的数据发生变化时，监听器会被触发，从而执行相应的回调方法，重新获取节点数据并打印出来。
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;public class ZooKeeperWatcherExample implements Watcher {// ZooKeeper 服务器地址private static final String ZOOKEEPER_ADDRESS = "localhost:2181";// 会话超时时间private static final int SESSION_TIMEOUT = 3000;// 节点路径private static final String NODE_PATH = "/example";// ZooKeeper 客户端private ZooKeeper zooKeeper;// 构造方法，在初始化时连接 ZooKeeper 服务器public ZooKeeperWatcherExample() {try {// 连接 ZooKeeper 服务器this.zooKeeper = new ZooKeeper(ZOOKEEPER_ADDRESS, SESSION_TIMEOUT, this);} catch (Exception e) {e.printStackTrace();}}// Watcher 接口的回调方法，处理节点状态变化事件@Overridepublic void process(WatchedEvent event) {// 节点状态变化事件if (event.getType() == Event.EventType.NodeDataChanged ||event.getType() == Event.EventType.NodeCreated ||event.getType() == Event.EventType.NodeDeleted) {// 重新获取节点数据，并注册监听器getNodeData();}}// 获取节点数据，并注册监听器private void getNodeData() {try {// 获取节点数据和状态信息Stat stat = new Stat();byte[] data = zooKeeper.getData(NODE_PATH, this, stat);String nodeData = new String(data);System.out.println("Node data: " + nodeData);// 注册监听器zooKeeper.exists(NODE_PATH, this);} catch (Exception e) {e.printStackTrace();}}// 设置节点数据public void setNodeData(String data) {try {// 检查节点是否存在Stat stat = zooKeeper.exists(NODE_PATH, false);if (stat != null) {// 设置节点数据zooKeeper.setData(NODE_PATH, data.getBytes(), stat.getVersion());} else {System.out.println("Node doesn't exist");}} catch (Exception e) {e.printStackTrace();}}// 主方法，用于演示public static void main(String[] args) throws InterruptedException {// 创建示例对象ZooKeeperWatcherExample example = new ZooKeeperWatcherExample();// 创建示例节点try {example.zooKeeper.create(NODE_PATH, "Initial data".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);} catch (Exception e) {e.printStackTrace();}// 设置节点数据example.setNodeData("Updated data");// 等待一段时间，观察节点状态变化Thread.sleep(5000);// 关闭 ZooKeeper 连接try {example.zooKeeper.close();} catch (Exception e) {e.printStackTrace();}}
}
