Zookeeper如何网络通信+监听？看一看Watch机制

电玩女神 2023-10-04 22:08 41阅读 0赞

## Zookeeper Watch机制 ##

Watcher是一种简单的机制，使客户端得到关于ZooKeeper集合中的更改的通知。 客户端可以在读取特定znode时设置Watcher。Watcher会向注册的客户端发送任何znode(客户端注册表)更改的通知。

### 1. 概述 ###

ZooKeeper Watch 机制是指，客户端在所有的读命令上告知服务端：这个节点或者子节点变化时通知我，具体来说，支持的写操作有：

*  getData
 *  getChildren
 *  exists

例如，我们在命令行可以输入 get -w /foo，其中 -w 参数就是用于告知 ZooKeeper 服务端，当前客户端想在 /foo 节点上设置一个监听器。

ZooKeeper Watch 机制的两个细节：

*  wactch 是一次性触发的（除了永久递归 watch），如果客户端如果在一个 watch 通知后继续收到相同节点的 watch 通知，那么必须再次注册 watch 一次；
 *  服务端发给客户端的 watch 通知并不包含具体的节点数据，其起到的作用非常存粹：告知客户端其关注的节点发生了 watch 事件；

本篇博客在客户端角度，从底层出发，看一下Zookeeper Watch机制。开始之前，先思考一下以下疑问，带着这些问题进行Zookeeper客户端的学习。

*  Zookeeper 客户端如何进行网络请求
 *  Zookeeper 如何处理同步和异步请求
 *  Zookeeper如何注册和触发Watcher
 *  我们常用的ZkClient又做了什么？

### 2. 客户端网络IO模型 ###

Copy From [ZooKeeper客户端源码解读（网络I/O）][ZooKeeper_I_O]

#### 2.1 整体结构图 ####

ClientCnxnSocket 封装了底层Socket通信层, ClientCnxnSocket整体结构如图所示：

![47e92284256f4ff1e6b0f93217cc6143.png][]

#### 2.2 Packet ####

Packet是ClientCnxn内部定义的一个对协议层的封装，作为ZooKeeper\]中请求与响应的载体。

![4505fbc9ab3e87eaf79942939f785c57.png][]

![73c6ac6b6deceecd4e832dd495ec0077.png][]

从上图可以看出，Packet中包含了请求头、响应头、请求体、响应体、节点路径和注册的Watcher等信息。

#### 2.3 SenderThread ####

2.3.1 基本概念

SendThread是客户端ClientCnxn内部一个核心的I/O调度线程，用于管理客户端和服务端之间的所有网络I/O操作。在ZooKeeper客户端的实际运行过程中

*  SendThead维护了客户端和服务端之间的会话生命周期，其通过在一定的周期频率内向服务器发送一个PING包来实现心跳检测，同时，在会话周期内，如果客户端和服务端之间出现TCP连接断开的情况，那么就会自动而且透明化完成重连操作。
 *  另一方面，SendThread管理了客户端所有的请求发送和响应接收操作，其将上层客户端API操作转换成相应的请求协议并发送到服务端，并完成对同步调用的返回和异步调用的回调。
 *  同时，SendThread还负责将来自服务端的事件传递给EventThread去处理。

Sender进程就一直尝试与Zookeeper服务器进行交互：

//org.apache.zookeeper.ClientCnxn.SendThread
    
    @Override
    public void run() {
       // ...
       while (state.isAlive()) {
           clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this);
       }
       //...
    }

// org.apache.zookeeper.ClientCnxnSocketNIO#doTransport
     void doTransport(...) {
             ...
            //监听Selector,对读和写进行操作
            for (SelectionKey k : selected) {
                ...
                if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
                    //doIO
                    doIO(pendingQueue, cnxn);
                }
              ...
            }
    
        }
    
    void doIO(Queue<Packet> pendingQueue, ClientCnxn cnxn) throws InterruptedException, IOException {

[ZooKeeper_I_O]: https://blog.csdn.net/m0_37607945/article/details/105197680?utm_medium=distribute.pc_relevant.none-task-blog-2~default~baidujs_title~default-0-105197680-blog-79314974.pc_relevant_antiscanv4&spm=1001.2101.3001.4242.1&utm_relevant_index=3
[47e92284256f4ff1e6b0f93217cc6143.png]: https://img-blog.csdnimg.cn/img_convert/47e92284256f4ff1e6b0f93217cc6143.png
[4505fbc9ab3e87eaf79942939f785c57.png]: https://img-blog.csdnimg.cn/img_convert/4505fbc9ab3e87eaf79942939f785c57.png
[73c6ac6b6deceecd4e832dd495ec0077.png]: https://img-blog.csdnimg.cn/img_convert/73c6ac6b6deceecd4e832dd495ec0077.png