zookeeper的watcher机制

朴灿烈づ我的快乐病毒、 2023-01-17 07:41 131阅读 0赞

ZooKeeper 提供了分布式数据的发布/订阅功能。一个典型的发布/阅模型系统定义了一种一对多的订阅关系，能够让多个订阅者同时监听某一个主题对象，当这个主题对象自身状态变化时，会通知所有订阅者，使它们能够做出相应的处理。在 ZooKeeper 中，引入了 Watcher 机制来实现这种分布式的通知功能。 ZooKeeper 允许客户端向服务端注册一个 Watcher 监听，当服务端的一些指定事件触发了这个 Watcher，那么就会向指定客户端发送一个事件通知来实现分布式的通知功能。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdjaGVuZ21pbmcx_size_16_color_FFFFFF_t_70]  
简单来说客户端在向zookeeper服务器注册watcher的同时，会将watcher对象存储在客户端的WatchManager中，当zookeeper服务器触发watcher事件后，会向客户端发送通知，客户端线程从WatchManager中拉取对应的watcher对象来执行回调逻辑。

### watcher的EventType ###

EventType代表了watcher的事件类型，枚举如下：

None(-1), // 客户端与服务器成功建立会话
    NodeCreated(1), // watcher监听的节点被创建
    NodeDeleted(2), // watcher监听的节点被删除
    NodeDataChanged(3), // watcher监听的节点数据内容发生变化
    NodeChildrenChanged(4); // watcher监听的节点的子节点列表内容发生变化

### watcher的KeeperState ###

KeeperState代表了watcher的通知状态，枚举如下：

Disconnected(0), // 客户端和服务端处于连接断开状态
    SyncConnected(3), // 客户端和服务端处于连接状态
    AuthFailed(4), // 权限检查失败
    ConnectedReadOnly(5), // 客户端连接到一个只读的服务端
    SaslAuthenticated(6), // 通知客户端他们是SASL认证
    Expired(-112); // 会话超时

### 回调方法process() ###

process方法是watcher接口中的一个回调方法，当zookeeper向客户端发送一个watcher事件通知时，客户端就会对响应的process方法进行回调，从而实现对事件的处理。

void process(WatchedEvent event)

process方法的定义很简单，我们再看一下参数`WatchedEvent`

public class WatchedEvent { 
        private final KeeperState keeperState;
        private final EventType eventType;
        private String path;
    }

不难理解，zookeeper通过WatchedEvent对象封装服务端事件并传递给watcher，从而方便回调方法process对服务端事件处理。举个例子：当`/zk-book` 这个节点的数据发生变更时，服务端会发送给客户端一个"ZNode 数据内容变更"事件，客户端只能够接收到如下信息

KeeperState: SyncConnected 
    EventType; NodeDataChanged 
    Path: /zk-book

从上面展示的信息中，我们可以看到，客户端无法直接从该事件中获取到对应数据节点的原始数据内容以及变更后的新数据内容，而是需要客户端再次主动去重新获取数据——这也是 ZooKeeper Watcher 机制的一个非常重要的特性。

接下来我们详细看一下ZooKeeper Watcher机制的实现。

### watcher的工作机制 ###

zookeeper的watcher机制，总的来说可以包括以下三个过程：

*  客户端注册watcher
 *  服务端处理watcher
 *  客户端回调watcher

其内部各组件的关系如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdjaGVuZ21pbmcx_size_16_color_FFFFFF_t_70 1]

##### 客户端注册watcher #####

我们在创建一个zookeeper客户端对象实例时，可以向构造方法中传入一个默认的watcher，源码如下：

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException { 
            this(connectString, sessionTimeout, watcher, false);
    }

这个watcher将作为整个zookeeper会话期间的默认watcher，会被一直保存在客户端ZKWatchManager的defaultWatcher中。

无论是 ZooKeeeper 命令行客户端，还是 ZooKeeper 类本身，都最终通过 ZooKeeper 实例的相关方法完成注册，即：getData()、getChildren() 以及 exist() 方法。例如，getData() 有多个重载方法，如下：

public void getData(String path, boolean watch, DataCallback cb, Object ctx) { 
            getData(path, getDefaultWatcher(watch), cb, ctx);
        }
    
    public byte[] getData(String path, boolean watch, Stat stat) throws KeeperException, InterruptedException { 
            return getData(path, getDefaultWatcher(watch), stat);
        }
    
    public void getData(final String path, Watcher watcher, DataCallback cb, Object ctx) { 
        //省略具体实现逻辑
        }
    
    public byte[] getData(final String path, Watcher watcher, Stat stat) throws KeeperException, InterruptedException { 
        //省略具体实现逻辑
    }

客户端注册watcher的逻辑如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdjaGVuZ21pbmcx_size_16_color_FFFFFF_t_70_pic_center]

##### 服务端处理watcher #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdjaGVuZ21pbmcx_size_16_color_FFFFFF_t_70 2]  
上面是服务端处理watcher的时序图

*  服务端收到来自客户端的请求之后，在FinalRequest Processor.processRequest()中会判断当前请求是否需要注册Watcher
 *  检查请求是否有设置 wacth 参数，并根据是否存在来决定如何从 ZKDatabase 实例上获取 Node 数据
    
     *  如果有设置，那么通过向 ZKDatabase 实例的 getData() 方法传入：path、stat 以及当前 ServerCnxn 实例(ServerCnxn 实例对应于当前 ZooKeeeper 服务端与客户端的连接)
     *  如果没有设置，那么通过向 ZKDatabase 实例的 getData() 方法传入：path、stat 以及 null
 *  ZKDatabase.getData(String path, Stat stat, Watcher watcher) 方法的执行逻辑为
    
     *  检查入口参数 path 对应的节点是否存在，如果不存在，那么直接抛出异常退出，否则，继续下一步
     *  从节点读取相关数据复制到 stat 实例
     *  利用 WatchManager.addWatch(String path, Watcher watcher) 方法来完成 watch 的注册（真正是否注册取决于 watcher 是否为 null）
     *  得到节点的数据 data并返回
 *  最后 ZKDatabase 实例实际上也没有完成 Watcher 的注册工作，而是将 Watcher 注册任务交给了 WatcherManager 来实现。WatcherManager.addWatch(String path, Watcher watcher) 方法的运行逻辑如下
    
     *  如果存在 path 对应的 `Set<Watcher> list`，那么就将当前 Watcher 添加到这个 list 中去；如果没有，那么新建一个，再添加
     *  如果存在当前 Watcher 对应的 `Set<String>（String 的语义为 path）list`，那么就将当前 path 加入到这个 list 中取；如果没有，那么新建一个，再添加
     *  watcher 模式注册：如果默认模式，那么没有必要注册，我们需要确保其不在 watcherModeManager 实例的 watcherModes（hashMap）字段中注册，非默认模式则要确保注册
 *  Watch 注册成功后，我们达到了这样这样的效果
    
     *  通过给 WatchManager 实例提供 path，我们能够得到所有监听此 path 的 Watcher 实例
     *  通过给 WatchManager 实例提供 Watcher，我们能够得到该 Watcher 实例正监听着的所有节点对应的 path

##### 客户端回调watcher #####

Process 方法是 Watcher 接口中的一个回调方法，当 ZooKeeper 向客户端发送一个 Watcher 事件通知时，客户单就会对相应的 process 方法进行回调，从而实现对事件的处理。这个回调方法的定义非常简单，我们重点看下方法的参数定义WatchedEvent。WatchedEvent 包含了每一个事件的三个基本属性：通知状态（keeperState）、事件类型（eventType）和节点路径（path）。ZooKeeper 使用 WatchedEvent 对象来封装服务端事件并传递给 Watcher，从而方便回调方法 process 对服务端事件进行处理。

if (replyHdr.getXid() == -1) { 
                    if (ClientCnxn.LOG.isDebugEnabled()) { 
                        ClientCnxn.LOG.debug("Got notification sessionid:0x" + Long.toHexString(ClientCnxn.this.sessionId));
                    }
    
                    WatcherEvent event = new WatcherEvent();
                    event.deserialize(bbia, "response");
                    if (ClientCnxn.this.chrootPath != null) { 
                        String serverPath = event.getPath();
                        if (serverPath.compareTo(ClientCnxn.this.chrootPath) == 0) { 
                            event.setPath("/");
                        } else if (serverPath.length() > ClientCnxn.this.chrootPath.length()) { 
                            event.setPath(serverPath.substring(ClientCnxn.this.chrootPath.length()));
                        } else { 
                            ClientCnxn.LOG.warn("Got server path " + event.getPath() + " which is too short for chroot path " + ClientCnxn.this.chrootPath);
                        }
                    }
    
                    WatchedEvent we = new WatchedEvent(event);
                    if (ClientCnxn.LOG.isDebugEnabled()) { 
                        ClientCnxn.LOG.debug("Got " + we + " for sessionid 0x" + Long.toHexString(ClientCnxn.this.sessionId));
                    }
    
                    ClientCnxn.this.eventThread.queueEvent(we);
                }

### ZooKeeper Watcher机制的特性总结 ###

*  一次性

无论是服务端还是客户端，一旦一个 Watcher 被触发，ZooKeeper都会将其从相应的存储中移除。因此，开发人员在 Watcher 的使用上要记住的一点是需要反复注册。这样的设计有效地减轻了服务端的压力。试想，如果注册一个 Watcher 之后一直有效，那么，针对那些更新非常频繁的节点，服务端会不断地向客户端发送事件通知，这无论对于网络还是服务端性能的影响都非常大。需要注意的是：ZooKeeper 在 3.6.0 版本中引入了永久 Watcher 机制，利用这个机制可以避免反复注册 Watcher。

*  客户端串行执行

客户端 Watcher 回调的过程是一个串行同步的过程，这为我们保证了顺序，同时，需要开发人员注意的一点是，千万不要因为一个 Watcher 的处理逻辑影响了整个客户端的 Watcher 回调。

*  轻量

WatchedEvent 是 ZooKeeper 整个 Watcher 通知机制的最小通知单元，这个数据结构中只包含三部分内容:通知状态、事件类型和节点路径。也就是说，Watcher 通知非常简单，只会告诉客户端发生了事件，而不会说明事件的具体内容。例如针对 NodeDataChanged 事件，ZooKeeper 的 Watcher 只会通知客户端指定数据节点的数据内容发生了变更，而对于原始数据以及变更后的新数据都无法从这个事件中直接获取到，而是需要客户端主动重新去获取数据。这也是ZooKeeper 的 Watcher机制的一个非常重要的特性。

另外，客户端向服务端注册 Watcher 的时候，并不会把客户端真实的 Watcher 对象传递到服务端，仅仅只是在客户端请求中使用 boolean 类型属性进行了标记，同时服务端也仅仅只是保存了当前连接的 ServerCnxn 对象。如此轻量的 Watcher 机制设计，在网络开销和服务端内存开销上都是非常廉价的。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdjaGVuZ21pbmcx_size_16_color_FFFFFF_t_70]: /images/20221021/7c6b01514aa04905884807c133544eed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdjaGVuZ21pbmcx_size_16_color_FFFFFF_t_70 1]: /images/20221021/de0e42483297438abe02a5d2664f9c8e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdjaGVuZ21pbmcx_size_16_color_FFFFFF_t_70_pic_center]: /images/20221021/b6cd987025bf4a929a385ac3c14fb4c0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdjaGVuZ21pbmcx_size_16_color_FFFFFF_t_70 2]: /images/20221021/2b34f177645a4272888dda8a7a9fccf7.png