8.3 Watcher 机制源码分析

文档摘要

8.3 Watcher 机制源码分析 8.3 Zookeeper Watcher 机制源码分析 8.3.1 Watcher 机制概述在分布式协调服务Zookeeper中，Watcher机制是其核心特性之一，它允许客户端在关注的ZNode节点上设置监听器（Watcher），一旦被监听的ZNode节点发生特定事件（例如数据变更、节点创建、节点删除、子节点变更），Zookeeper服务端会主动通知所有在该节点上注册了Watcher的客户端。 Watcher 的关键特性: 一次性触发: Watcher 机制是一次性触发的。一旦 Watcher 被触发，它就会被移除，如果客户端需要持续监听，必须重新注册 Watcher。

8.3 Watcher 机制源码分析

8.3 Zookeeper Watcher 机制源码分析

8.3.1 Watcher 机制概述

在分布式协调服务Zookeeper中，Watcher机制是其核心特性之一，它允许客户端在关注的ZNode节点上设置监听器（Watcher），一旦被监听的ZNode节点发生特定事件（例如数据变更、节点创建、节点删除、子节点变更），Zookeeper服务端会主动通知所有在该节点上注册了Watcher的客户端。

Watcher 的关键特性:

一次性触发: Watcher 机制是一次性触发的。一旦 Watcher 被触发，它就会被移除，如果客户端需要持续监听，必须重新注册 Watcher。
异步通知: Watcher 的通知是异步的，客户端不需要轮询服务端，服务端在事件发生时主动推送通知给客户端。
轻量级: Watcher 通知是非常轻量级的，服务端只发送事件类型和节点路径等少量信息给客户端，不会传输大量数据。
顺序保证: Zookeeper 保证 Watcher 事件的顺序性，即客户端接收到的 Watcher 事件顺序与服务端事件发生的顺序一致。
可靠性: Zookeeper 保证 Watcher 事件的可靠传递，除非客户端与服务端断开连接，否则 Watcher 事件一定会被客户端接收到。

Watcher 的应用场景:

配置管理: 客户端可以 Watch 配置文件节点，当配置发生变更时，客户端能够及时收到通知并更新本地配置。
分布式锁: Watcher 可以用于实现分布式锁的释放通知，当持有锁的客户端释放锁时，可以通过 Watcher 通知等待锁的客户端。
集群管理: 客户端可以 Watch 集群节点，当集群成员发生变化时，客户端能够及时感知并进行相应的处理。

8.3.2 Watcher 机制源码分析

Zookeeper Watcher 机制的源码分析主要围绕以下几个核心流程展开：

Watcher 的注册: 客户端如何在服务端注册 Watcher。
Watcher 的存储: 服务端如何存储和管理注册的 Watcher。
Watcher 的触发: 服务端在事件发生时如何触发 Watcher。
Watcher 的通知: 服务端如何将 Watcher 事件通知给客户端。
客户端 Watcher 处理: 客户端如何接收和处理 Watcher 事件。

下面我们将逐一深入源码进行分析。

8.3.2.1 Watcher 的注册

Watcher 的注册通常发生在客户端调用 Zookeeper API 时，例如 getData(), exists(), getChildren() 等方法，并在这些方法中设置 watch 参数为 true。

客户端注册流程:

客户端发起请求: 客户端调用 ZooKeeper 对象的 getData(), exists(), getChildren() 等方法，并设置 watch=true。同时，客户端需要实现 Watcher 接口，并将其作为参数传递给上述方法。
构建 Watcher 注册请求: 客户端在 ClientCnxn 中将 Watcher 注册请求封装成一个请求对象，并发送给服务端。请求对象中包含要 Watch 的 ZNode 路径、Watcher 类型（GetData, Exists, GetChildren）以及客户端 Session ID 等信息。
服务端接收请求: 服务端 NIOServerCnxn 接收到客户端的 Watcher 注册请求。
服务端处理请求: 服务端 ZooKeeperServer 的 processPacket() 方法接收请求，并根据请求类型调用相应的处理器进行处理。对于 Watcher 注册请求，会调用 GetDataRequestProcessor, ExistsRequestProcessor, GetChildrenRequestProcessor 等处理器。
服务端存储 Watcher: 请求处理器在处理请求时，会将 Watcher 信息存储到 DataTree 对象的 watchTable 或 childWatchTable 中。watchTable 用于存储数据 Watcher，childWatchTable 用于存储子节点 Watcher。

代码示例 (客户端 Watcher 注册):


import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.List;
public class WatcherDemo {
    private static final String CONNECT_STRING = "localhost:2181";
    private static final int SESSION_TIMEOUT = 5000;
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        zooKeeper = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, new MyWatcher());
        // 确保连接建立
        while (zooKeeper.getState() != ZooKeeper.States.CONNECTED) {
            Thread.sleep(100);
        }
        String path = "/test-watcher";
        // 创建节点
        if (zooKeeper.exists(path, false) == null) {
            zooKeeper.create(path, "initial data".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        // 注册 Data Watcher
        zooKeeper.getData(path, true, new GetDataWatcher(), null);
        System.out.println("Data Watcher registered on: " + path);
        // 注册 Child Watcher
        zooKeeper.getChildren(path, true, new GetChildrenWatcher(), null);
        System.out.println("Child Watcher registered on: " + path);
        Thread.sleep(Long.MAX_VALUE); // 保持程序运行，等待 Watcher 事件
    }
    static class MyWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("Default Watcher Event: " + event);
        }
    }
    static class GetDataWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("GetData Watcher Event: " + event);
            if (event.getType() == Event.EventType.NodeDataChanged) {
                try {
                    // 重新获取数据并再次注册 Watcher (如果需要持续监听)
                    byte[] data = zooKeeper.getData(event.getPath(), true, this, null);
                    System.out.println("Node Data Changed, New Data: " + new String(data));
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    static class GetChildrenWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("GetChildren Watcher Event: " + event);
            if (event.getType() == Event.EventType.NodeChildrenChanged) {
                try {
                    // 重新获取子节点列表并再次注册 Watcher (如果需要持续监听)
                    List<String> children = zooKeeper.getChildren(event.getPath(), true, this, null);
                    System.out.println("Node Children Changed, New Children: " + children);
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

Mermaid 图 (Watcher 注册流程):

8.3.2.2 Watcher 的存储

服务端使用 DataTree 类来存储和管理 Watcher。DataTree 是 Zookeeper 服务端内存数据结构的核心，它以树形结构存储了 Zookeeper 的所有节点数据，同时也负责 Watcher 的管理。

DataTree 中用于存储 Watcher 的主要数据结构是：

watchTable (HashMap<String, Set>): 用于存储 数据 Watcher。Key 是 ZNode 路径，Value 是一个 Set 集合，存储了所有在该路径上注册了数据 Watcher 的 ServerCnxn 对象。ServerCnxn 代表了客户端连接。
childWatchTable (HashMap<String, Set>): 用于存储 子节点 Watcher。Key 是 ZNode 路径，Value 也是一个 Set 集合，存储了所有在该路径上注册了子节点 Watcher 的 ServerCnxn 对象。

Watcher 存储结构:

当客户端注册 Watcher 时，服务端会将客户端对应的 ServerCnxn 对象添加到 watchTable 或 childWatchTable 相应 ZNode 路径的 Set 集合中。使用 Set 集合是为了避免同一个客户端在同一个 ZNode 路径上重复注册相同的 Watcher。

8.3.2.3 Watcher 的触发

Watcher 的触发发生在服务端检测到被 Watch 的 ZNode 节点发生了特定事件时。可能触发 Watcher 的事件类型包括：

NodeCreated: 节点创建
NodeDeleted: 节点删除
NodeDataChanged: 节点数据变更
NodeChildrenChanged: 子节点列表变更

Watcher 触发流程:

事件发生: 服务端在处理客户端的写请求（例如 create(), delete(), setData(), create() 子节点等操作）时，会检测是否会触发 Watcher 事件。
检测 Watcher: 在 DataTree 中，当节点数据或子节点列表发生变更时，会根据事件类型和 ZNode 路径，查找 watchTable 或 childWatchTable 中是否存在注册的 Watcher。
获取 Watcher 列表: 根据 ZNode 路径从 watchTable 或 childWatchTable 中获取注册的 Watcher (ServerCnxn) 集合。
创建 Watcher 事件: 为每个 Watcher 创建一个 WatchedEvent 对象，包含事件类型、事件状态 (通常为 SyncConnected) 和 ZNode 路径。
移除 Watcher (一次性触发): 在触发 Watcher 之前，服务端会从 watchTable 或 childWatchTable 中移除已经触发的 Watcher。这保证了 Watcher 的一次性触发特性。
将 Watcher 事件放入队列: 将创建的 WatchedEvent 对象放入每个 ServerCnxn 对象关联的发送队列 outgoingQueues 中，等待发送给客户端。

代码示例 (服务端 Watcher 触发 - DataTree.deleteNode() 简化代码):


// DataTree.java (简化代码)
public boolean deleteNode(String path, int version) throws KeeperException.NoNodeException, KeeperException.BadVersionException, KeeperException.NotEmptyException {
    // ... 省略删除节点逻辑 ...
    // 触发 Data Watcher
    Set<ServerCnxn> watchers = watchTable.remove(path); // 移除并获取 Watcher
    if (watchers != null) {
        for (ServerCnxn watcherCnxn : watchers) {
            // 创建 NodeDeleted 事件
            WatchedEvent event = new WatchedEvent(Watcher.Event.EventType.NodeDeleted, Watcher.Event.KeeperState.SyncConnected, path);
            watcherCnxn.sendResponse(null, watcherCnxn.getZxid(), 0, new WatchedEventWrapper(event)); // 将事件放入发送队列
        }
    }
    // 触发 Child Watcher (如果删除的是父节点，需要通知子节点的 Child Watcher)
    String parentPath = PathUtils.parentPath(path);
    Set<ServerCnxn> childWatchers = childWatchTable.get(parentPath);
    if (childWatchers != null) {
        for (ServerCnxn watcherCnxn : childWatchers) {
            // 创建 NodeChildrenChanged 事件
            WatchedEvent event = new WatchedEvent(Watcher.Event.EventType.NodeChildrenChanged, Watcher.Event.KeeperState.SyncConnected, parentPath);
            watcherCnxn.sendResponse(null, watcherCnxn.getZxid(), 0, new WatchedEventWrapper(event)); // 将事件放入发送队列
        }
    }
    return true;
}

Mermaid 图 (Watcher 触发流程):

8.3.2.4 Watcher 的通知

Watcher 事件被放入 ServerCnxn 的发送队列 outgoingQueues 后，由 SendThread 线程负责将队列中的事件发送给客户端。

Watcher 通知流程:

SendThread 线程: 每个 ServerCnxn 都有一个关联的 SendThread 线程，负责从 outgoingQueues 队列中取出消息并发送给客户端。
序列化 Watcher 事件: SendThread 将 WatchedEvent 对象序列化成网络传输的字节流。
网络发送: SendThread 通过 Socket 连接将序列化后的 Watcher 事件发送给客户端。
客户端接收: 客户端 ClientCnxn 接收到服务端发送的 Watcher 事件数据。

代码示例 (服务端 Watcher 通知 - SendThread.run() 简化代码):


// SendThread.java (简化代码)
public void run() {
    while (running) {
        try {
            // ... 省略其他消息处理 ...
            OutgoingQueueEntry entry = outgoingQueue.take(); // 从发送队列中取出消息
            if (entry instanceof WatchedEventWrapper) {
                WatchedEventWrapper eventWrapper = (WatchedEventWrapper) entry;
                // 序列化 Watcher 事件
                ByteBuffer bb = eventWrapper.getBB();
                // 发送数据
                sock.getChannel().write(bb);
                // ... 省略发送完成后的处理 ...
            }
        } catch (InterruptedException e) {
            // ... 省略异常处理 ...
        } catch (IOException e) {
            // ... 省略 IO 异常处理 ...
        }
    }
}

Mermaid 图 (Watcher 通知流程):

8.3.2.5 客户端 Watcher 处理

客户端 ClientCnxn 接收到服务端发送的 Watcher 事件后，需要将事件传递给客户端注册的 Watcher 对象进行处理。

客户端 Watcher 处理流程:

ClientCnxn 接收事件: 客户端 ClientCnxn 从 Socket 连接中读取数据，反序列化得到 WatchedEvent 对象。
获取默认 Watcher: ClientCnxn 获取在创建 ZooKeeper 对象时注册的默认 Watcher。
调用 Watcher.process() 方法: ClientCnxn 调用默认 Watcher 的 process(WatchedEvent event) 方法，将接收到的 WatchedEvent 对象作为参数传递给该方法。
客户端业务逻辑处理: 在 Watcher.process() 方法中，客户端可以根据 WatchedEvent 的事件类型和 ZNode 路径，执行相应的业务逻辑，例如更新本地缓存、重新获取配置、触发分布式锁释放等。
重新注册 Watcher (如果需要持续监听): 由于 Watcher 是一次性触发的，如果客户端需要持续监听某个 ZNode 节点的事件，需要在 Watcher.process() 方法中重新注册 Watcher。

代码示例 (客户端 Watcher 处理 - ClientCnxn.readResponse() 简化代码):


// ClientCnxn.java (简化代码)
void readResponse(ByteBuffer incomingBuffer) throws IOException {
    // ... 省略读取和反序列化逻辑 ...
    ReplyHeader header = new ReplyHeader();
    ByteBufferInputStream bbis = new ByteBufferInputStream(incomingBuffer);
    BinaryInputArchive archive = BinaryInputArchive.getArchive(bbis);
    header.deserialize(archive, "header");
    if (header.getType() == ZooDefs.OpCode.WATCH_EVENT) {
        // 处理 Watcher 事件
        WatcherEvent event = new WatcherEvent();
        event.deserialize(archive, "response");
        WatchedEvent watchedEvent = event.getWatchedEvent();
        // 获取默认 Watcher
        Watcher defaultWatcher = zooKeeper.getDefaultWatcher();
        if (defaultWatcher != null) {
            // 调用 Watcher.process() 方法
            defaultWatcher.process(watchedEvent);
        }
    }
    // ... 省略其他响应处理 ...
}

Mermaid 图 (客户端 Watcher 处理流程):

8.3.3 Watcher 机制代码实践

在实际开发中，我们经常需要使用 Watcher 机制来实现各种分布式协调功能。以下是一些常见的代码实践示例：

1. 配置中心:


public class ConfigCenter {
    private ZooKeeper zooKeeper;
    private String configPath = "/config";
    private String configData;
    public ConfigCenter(String connectString) throws IOException, InterruptedException, KeeperException {
        zooKeeper = new ZooKeeper(connectString, 5000, new ConfigWatcher());
        // 确保连接建立
        while (zooKeeper.getState() != ZooKeeper.States.CONNECTED) {
            Thread.sleep(100);
        }
        ensureConfigNodeExists();
        loadConfig();
    }
    private void ensureConfigNodeExists() throws KeeperException, InterruptedException {
        if (zooKeeper.exists(configPath, false) == null) {
            zooKeeper.create(configPath, "default config".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }
    public String getConfig() {
        return configData;
    }
    private void loadConfig() throws KeeperException, InterruptedException {
        byte[] data = zooKeeper.getData(configPath, true, new ConfigWatcher(), null);
        configData = new String(data);
        System.out.println("Config loaded: " + configData);
    }
    class ConfigWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            if (event.getType() == Event.EventType.NodeDataChanged && event.getPath().equals(configPath)) {
                System.out.println("Config node changed, reloading config...");
                try {
                    loadConfig(); // 重新加载配置
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        ConfigCenter configCenter = new ConfigCenter("localhost:2181");
        System.out.println("Current config: " + configCenter.getConfig());
        Thread.sleep(Long.MAX_VALUE); // 保持程序运行
    }
}

2. 分布式锁 (基于 Watcher 实现释放通知):


public class DistributedLock {
    private ZooKeeper zooKeeper;
    private String lockPath = "/mylock";
    private String currentLockNode;
    public DistributedLock(String connectString) throws IOException, InterruptedException, KeeperException {
        zooKeeper = new ZooKeeper(connectString, 5000, new DefaultWatcher());
        // 确保连接建立
        while (zooKeeper.getState() != ZooKeeper.States.CONNECTED) {
            Thread.sleep(100);
        }
        ensureLockPathExists();
    }
    private void ensureLockPathExists() throws KeeperException, InterruptedException {
        if (zooKeeper.exists(lockPath, false) == null) {
            zooKeeper.create(lockPath, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }
    public boolean acquireLock() throws KeeperException, InterruptedException {
        try {
            // 尝试创建临时顺序节点
            currentLockNode = zooKeeper.create(lockPath + "/lock-", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            List<String> children = zooKeeper.getChildren(lockPath, false);
            Collections.sort(children);
            if (currentLockNode.endsWith(children.get(0).substring(children.get(0).lastIndexOf('-') + 1))) {
                System.out.println("Acquired lock: " + currentLockNode);
                return true; // 获取到锁
            } else {
                // 没有获取到锁，需要 Watch 前一个节点
                String previousNode = getPreviousNode(children, currentLockNode);
                if (previousNode != null) {
                    Stat stat = zooKeeper.exists(lockPath + "/" + previousNode, new LockWatcher()); // Watch 前一个节点
                    if (stat != null) {
                        return false; // 等待 Watcher 通知
                    } else {
                        // 前一个节点已不存在，重新尝试获取锁
                        return acquireLock();
                    }
                }
            }
        } catch (KeeperException e) {
            if (e.code() == KeeperException.Code.NODEEXISTS) {
                return false; // 锁已被占用
            }
            throw e;
        }
        return false; // 默认返回 false
    }
    public void releaseLock() throws KeeperException, InterruptedException {
        if (currentLockNode != null) {
            zooKeeper.delete(currentLockNode, -1);
            currentLockNode = null;
            System.out.println("Released lock: " + currentLockNode);
        }
    }
    private String getPreviousNode(List<String> children, String currentLockNode) {
        Collections.sort(children);
        String currentNodeName = currentLockNode.substring(currentLockNode.lastIndexOf('/') + 1);
        int index = children.indexOf(currentNodeName);
        if (index > 0) {
            return children.get(index - 1);
        }
        return null; // 当前节点已经是最小节点
    }
    class LockWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            if (event.getType() == Event.EventType.NodeDeleted) {
                System.out.println("Previous node deleted, try to acquire lock again...");
                try {
                    acquireLock(); // 前一个节点删除，尝试重新获取锁
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    static class DefaultWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("Default Watcher Event: " + event);
        }
    }
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        DistributedLock lock = new DistributedLock("localhost:2181");
        if (lock.acquireLock()) {
            System.out.println("Do something with lock...");
            Thread.sleep(5000); // 模拟持有锁的操作
            lock.releaseLock();
        } else {
            System.out.println("Failed to acquire lock.");
        }
        Thread.sleep(Long.MAX_VALUE);
    }
}

8.3.4 总结

Zookeeper 的 Watcher 机制是构建分布式协调应用的重要基石。通过源码分析，我们深入了解了 Watcher 的注册、存储、触发和通知流程，以及客户端的处理方式。理解 Watcher 机制的原理和实现细节，能够帮助我们更好地使用 Zookeeper，并解决实际分布式系统中的各种协调问题。

关键要点回顾:

一次性触发: Watcher 只会被触发一次，需要持续监听需要重新注册。
异步通知: 服务端主动推送 Watcher 事件，客户端无需轮询。
数据结构: DataTree 中的 watchTable 和 childWatchTable 用于存储和管理 Watcher。
核心流程: 注册 -> 存储 -> 触发 -> 通知 -> 处理。
代码实践: Watcher 广泛应用于配置中心、分布式锁、集群管理等场景。

希望这篇文章能够帮助你深入理解 Zookeeper Watcher 机制的源码和应用。

作者与出处

原作者：灏天文库

来源：灏天文库

原文： https://aiknowledge.cn/collection/223-zookeeper%E5%88%86%E5%B8%83%E5%BC%8F%E5%8D%8F%E8%B0%83%E6%9C%8D%E5%8A%A1%E6%95%99%E7%A8%8B

整理：灏天文库整理

由灏天文库平台收录，内容或由平台用户上传，仅供学习交流