ZooKeeper系统模型之Watcher——数据变更的通知（工作机制）。

         ZooKeeper的Watcher机制，总的来说可以概括为以下三个过程：客户端注册Watcher、服务端的处理Watcher和客户端回调Watcher，其内部各组件之间的关系如下图所示。

客户端注册Watcher

         创建一个 ZooKeeper客户端对象实例时，可以向构造方法中传入一个默认的Watcher：

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher);
        这个Watcher将作为整个ZooKeeper会话期间的默认Watcher，会一直被保存在客户端ZKWatchManager的defaultWatcher中。另外，ZooKeeper客户端也可以通过getData、getChildren和exist三个接口来向ZooKeeper服务器注册Watcher，无论使用哪种方式，注册Watcher的工作原理都是一致的，这里我们以getData这个接口为例来说明。getData接口用于获取指定节点的数据内容，主要有两个方法：

public byte[] getData(String path, boolean watch, Stat stat)
public byte[] getData(final String path, Watcher watcher, Stat stat)
         在这两个接口上都可以进行Watcher的注册，第一个接口通过一个boolean参数来标识是否使用上文提到的默认Watcher来进行注册，具体的注册逻辑和第二个接口是一致的。

        在向getData接口注册Watcher后，客户端首先会对当前客户端请求request进行标记，将其设置为“使用Watcher监听”，同时会封装一个Watcher的注册信息WatchRegistration对象，用于暂时保存数据节点的路径和Watcher的对应关系，具体的逻辑代码如下：



        在ZooKeeper中，Packet可以被看作一个最小的通信协议单元，用于进行客户端与服务端之间的网络传输，任何需要传输的对象都需要包装成一个Packet对象。因此在ClientCnxn中WatchRegistration又会被封装到Packet中去，然后放入发送队列中等待客户端发送：



         随后，ZooKeeper客户端就会向服务端发送这个请求，同时等待请求的返回。完成请求发送后，会由客户端SendThread线程的readResponse方法负责接收到来自服务端的响应，finishPacket方法会从Packet中取出对应的Watcher并注册到ZKWatchManager中去：



        从上面的内容中，我们已经了解到客户端已经将Watcher暂时封装在了WatchRegistration对象，现在就需要从这个封装对象中再次提取出Watcher来：



         在register方法中，客户端会将之前暂时保存的Watcher对象转交给ZKWatchManager，并最终保存到dataWatches中去。ZKWatchManager.dataWatches是一个Map<String, Set<Watcher>>类型的数据结构，用于将数据节点的路径和Watcher对象进行一一映射后管理起来。整个客户端Watcher的注册流程如下图所示。



        我们发现，在极端情况下，客户端每调用一次getData()接口，就会注册上一个Watcher，那么这些Watcher实体都会随着客户端请求被发送到服务端去吗？

        答案是否定的。如果客户单注册的所有Watcher都被传递到服务端的话，那么服务端肯定会出现内存紧张或其他性能问题了，幸运的是，在ZooKeeper的设计中充分考虑到了这个问题。在上面的流程中，我们提到把WatchRegistration封装到了Packet对象中去，但事实上，在底层实际的网络传输序列化过程中，并没有将WatchRegistration对象完全的序列化到底层字节数组中去。为了证实这一点，我们可以看下Packet内部的序列化过程：



        从上面的代码片段中，我们可以看到，在Packet.createBB()方法中，ZooKeeper只会将requestHeader和request两个属性进行序列化，也就是说，尽管WatchRegistration被封装在了Packet中，但是并没有被序列化到底层字节数组中，因此也就不会进行网络传输了。

服务端处理Watcher

        上面主要讲解了客户端注册Watcher的过程，并且已经了解了最终客户端并不会将Watcher对象真正传递到服务端。那么，服务端究竟是如何完成客户端的Watcher注册，又是如何来处理这个Watcher的呢？下面将主要围绕这两个问题展开进行讲解。

ServerCnxn存储

        我们首先来看下服务器接收Watcher并将其存储起来的过程，如下图所示是ZooKeeper服务端处理Watcher的序列图。



        从上图中我们可以看到，服务端收到来自客户端的请求之后，在FinalRequestProcessor.processRequest()中会判断当前请求是否需要注册Watcher：



        从getData请求的处理逻辑中，我们可以看到，当getDataRequest.getWatch()为true的死后，ZooKeeper就认为当前客户端请求需要进行Watcher注册，于是就会将当前的ServerCnxn对象和数据节点路径传入getData方法中去。那么为什么要传入ServerCnxn呢？ServerCnxn是一个ZooKeeper客户端和服务器之间的连接接口，代表了一个客户端和服务器的连接。ServerCnxn接口的默认实现是NIOServerCnxn，同时从3.4.0版本开始，引入了基于Netty的实现：NettyServerCnxn。无论采用哪种实现方式，都实现了Watcher的process接口，因此我们可以把ServerCnxn看作是一个Watcher对象。数据节点的节点路径和ServerCnxn最终会被存储在WatchManager的watchTable和watch2Paths中。

        WatchManager是ZooKeeper服务端Watcher的管理者，其内部管理的watchTable和watch2Path2两个存储结构，分别从两个维度对Watcher进行存储。

watchTable是从数据节点路径的粒度来托管Watcher。
watch2Paths是从Watch的粒度来控制事件触发需要触发的数据节点。
        同时，WatchManager还负责Watcher事件的触发，并移除那些已经被触发的Watcher。注意，WatchManager只是一个统称，在服务端，DataTree中会托管两个WatchManager，分别是dataWatches和childWatches，分别对应数据变更Watcher和子节点变更Watcher。在本例中，因为是getData接口，因此最终会被存储在dataWatches中，其数据结构如下图所示。

Watch触发

        在上面的讲解中，我们了解了对于标记了Watcher注册的请求，ZooKeeper会将其对应的ServerCnxn存储到WatchManager中，下面我们来看看服务端是如何触发Watcher的。NodeDataChanged事件的触发条件是“Watcher监听的对应数据节点的数据内容发生变更”，其具体实现如下：





        在对指定节点进行数据更新后，通过调用WatchManager的triggerWatch方法来触发相关的事件：



        无论是dataWatches还是childWatches管理器，Watcher的触发逻辑都是一致的，基本步骤如下。

封装WatchedEvent。

首先将通知状态（KeeperState）、事件类型（EventType）以及节点路径（Path）封装成一个WatchedEvent对象。

查询Watcher。

根据数据节点的节点路径从watchTable中取出对应的Watcher。如果没有找到Watcher，说明没有任何客户端在该数据节点上注册过Watcher，直接退出。而如果找到了这个Watcher，会将其提取出来，同时会直接从watchTable和watch2Paths中将其删除——从这里我们也可以看出，Watcher在服务器是一次性的，及触发一次就失效了。

调用process方法来触发Watcher。

在这一步中，会逐个依次的调用从步骤2中找出的所有Watcher的process方法。那么这里的process方法究竟做了些什么呢？在上文中我们已经提到，对于需要注册Watcher的请求，ZooKeeper会把当前请求对应的ServerCnxn作为一个Watcher进行存储，因此，这里调用的process方法，事实上就是ServerCnxn的对应方法：



从上面的代码片段中，我们可以看出在process方法中，主要逻辑如下。

在请求头中标记“-1”，表明当前是一个通知。
将WatchedEvent包装成WatcherEvent，以便进行网络传输序列化。
向客户端发送该通知。
从上面几个步骤中可以看到，ServerCnxn的process方法中的逻辑非常简单，本质上并不是处理客户端Watcher真正的业务逻辑，而是借助当前客户端连接的ServerCnxn对象来实现对客户端的WatchedEvent传递，真正的客户端Watcher回调与业务逻辑执行都在客户端。

客户端回调Watcher

        上面我们已经讲解了服务端是如何进行Watcher触发的，并且知道了最终服务端会通过使用ServerCnxn对应的TCP连接来向客户端发送一个WatcherEvent事件，下面我们来看看客户单是如何处理这个事件的。

SendThread接收事件通知

        首先我们来看下ZooKeeper客户端是如何接收到这个客户端事件通知的：



        对于一个来自服务端的响应，客户端都是由SendThread.readResponse(ByteBuffer incomingBuffer)方法来统一进行处理的，如果响应头replyHdr中标识了XID为-1，表明这是一个通知类型的响应，对其的处理大体上分为以下4个主要步骤。

反序列化。

ZooKeeper客户端接收到请求后，首先会将字节流转换成WatcherEvent对象。

处理chrootPath。

如果客户端设置了chrootPath属性，那么需要对服务端传过来的完整的节点路径进行chrootPath处理，生成客户端的一个相对节点路径。例如客户端设置了chrootPath为/app1，那么针对服务端传过来的响应包含的节点路径为/app1/locks，经过chrootPath处理后，就会变成一个相对路径：/locks。

还原WatchedEvent。

process接口的参数定义是WatchedEvent，因此这里需要将WatcherEvent对象转换成WatchedEvent。

回调Watcher。

最后将WatchedEvent对象交给EventThread线程，在下一个轮询周期中进行Watcher回调。

EventThread处理事件通知

        在上面内容中我们讲到，服务端的Watcher事件通知，最终交给了EventThread线程来处理，现在我们就来看看EventThread的一些核心逻辑。EventThread线程是ZooKeeper客户端中专门用来处理服务端通知事件的线程，其数据结构如下图所示。



         在上文中，我们讲到SendThread接收到服务端的通知事件后，会通过调用EventThread.queueEvent方法将事件传给EventThread线程，其逻辑如下：





        queueEvent方法首先会根据该通知事件，从ZKWatchManager中取出所有相关的Watcher：



        客户端在识别出事件类型EventType后，会从相应的Watcher存储（即dataWatches、existWatches或childWatches中的一个或多个，本例中就是从dataWatches和existWatches两个存储中获取）中去除相应的Watcher。注意，此处使用的是remove接口，因此也表明了客户端的Watcher机制同样也是一次性的，即一旦被触发后，该Watcher就失效了。

        获取到相关的所有Watcher之后，会将其放入waitingEvents这个队列中去。WaitingEvents是一个待处理Watcher的队列，EventThread的run方法会不断对该队列进行处理：



        从上面的代码片段中我们可以看出，EventThread线程每次都会从waitingEvent队列中取出一个Watcher，并进行串行同步处理。注意，此处processEvent方法中Watcher才是之前客户端真正注册的Watcher，调用其process方法就可以实现Watcher的回调了。