我们前面提到，ZooKeeper提供了分布式数据的发布/订阅功能，一个典型的发布/订阅模型系统定义了一种一对多的订阅关系，能够让多个订阅者同时监听某一个主题对象，当这个主题对象自身状态变化时，会通知所有订阅者，使他们能够做出相应的处理。在ZooKeeper中，引入了Watchert机制来实现这种分布式的通知功能。ZooKeeper允许客户端向服务端注册一个Watcher监听，当服务端的一些事件触发了这个Watcher，那么就会向指定客户端发送一个事件通知来实现分布式的通知功能。

当客户端向ZooKeeper服务器注册Watcher的同时，会将Watcher对象存储在客户端的WatchManager中。当ZooKeeper服务器端触发Watcher事件后，会向客户端发送通知，客户端线程从WatchManager中取出对应的Watcher对象来执行回调逻辑。

Watcher接口

在ZooKeeper中，接口类Watcher用于表示一个标准的事件处理器，其定义了事件通知相关的逻辑，包含KeeperState和EventType两个枚举类，分别代表了通知状态和事件类型，同时定义了事件的回调方法：process（WatchedEvent event）。

Watcher事件

同一个事件类型在不同的通知状态中代表的含义有所不同，表7-3列举了常见的通知状态和事件类型。

表7-3 Watcher通知状态与事件类型一览

KeeperState	EventType	触发条件	说明
	None （-1）	客户端与服务端成功建立连接
SyncConnected （0）	NodeCreated （1）	Watcher监听的对应数据节点被创建
	NodeDeleted （2）	Watcher监听的对应数据节点被删除	此时客户端和服务器处于连接状态
	NodeDataChanged （3）	Watcher监听的对应数据节点的数据内容发生变更
	NodeChildChanged （4）	Wather监听的对应数据节点的子节点列表发生变更
Disconnected （0）	None （-1）	客户端与ZooKeeper服务器断开连接	此时客户端和服务器处于断开连接状态
Expired （-112）	Node （-1）	会话超时	此时客户端会话失效，通常同时也会受到SessionExpiredException异常
AuthFailed （4）	None （-1）	通常有两种情况，1：使用错误的schema进行权限检查 2：SASL权限检查失败	通常同时也会收到AuthFailedException异常

表7-3中列举了ZooKeeper中最常见的几个通知状态和事件类型。

回调方法process（）

process方法是Watcher接口中的一个回调方法，当ZooKeeper向客户端发送一个Watcher事件通知时，客户端就会对相应的process方法进行回调，从而实现对事件的处理。process方法的定义如下：

abstract public void process(WatchedEvent event);

这个回调方法的定义非常简单，我们重点看下方法的参数定义：WatchedEvent。

WatchedEvent包含了每一个事件的三个基本属性：通知状态（keeperState），事件类型（EventType）和节点路径（path），其数据结构如图7-5所示。ZooKeeper使用WatchedEvent对象来封装服务端事件并传递给Watcher，从而方便回调方法process对服务端事件进行处理。

提到WatchedEvent，不得不讲下WatcherEvent实体。笼统地讲，两者表示的是同一个事物，都是对一个服务端事件的封装。不同的是，WatchedEvent是一个逻辑事件，用于服务端和客户端程序执行过程中所需的逻辑对象，而WatcherEvent因为实现了序列化接口，因此可以用于网络传输。

服务端在生成WatchedEvent事件之后，会调用getWrapper方法将自己包装成一个可序列化的WatcherEvent事件，以便通过网络传输到客户端。客户端在接收到服务端的这个事件对象后，首先会将WatcherEvent还原成一个WatchedEvent事件，并传递给process方法处理，回调方法process根据入参就能够解析出完整的服务端事件了。

需要注意的一点是，无论是WatchedEvent还是WatcherEvent，其对ZooKeeper服务端事件的封装都是机及其简单的。举个例子来说，当/zk-book这个节点的数据发生变更时，服务端会发送给客户端一个“ZNode数据内容变更”事件，客户端只能够接收到如下信息：

keeperState：SyncConnected

EventType：NodeDataChanged

Path：/zk-book

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

工作机制

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

图7-7 Watcher相关UML

客户端注册Watch

在前面我们提到在创建一个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)

在这恋歌接口上都可以进行Watch的注册，第一个接口通过一个boolean参数来标识是否使用上文提到的默认Watcher来进行注册，具体的注册逻辑和第二个接口是一致的。

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

public Stat getDaTA(final String path,Watcher watcher, Stat stat){
     ....
     WatchRegistration wcb = null;
     if(watcher != null){
          wcb = new DataWatchRegistration(watcher,clientPath);
     }    
     ....
     request.setWatch(watch!=null);
     ReplyHeader r = cnxn.submitRequest(h,request,response,wcb);
     ...
}

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

Packet queuePacket(Request h,ReplyHeader r,Record request, Record response, AsyncCallback cb, String clientPath,String serverPath,Object ctx,WatchRegistration watchRegistration){
       Packet packet = null;
       ...
       synchronized(outgoingQueue){
        packet = new Packet(h,r,request,response,watchRegistration);
       ...
       outgoingQueue.add(packet);
       ...
}

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

private void vinishPacket(Packet p){
     if(p.watchRegistration != null){
          p.watchRegistration.register(p.replyHeader.getErr());
     }
     ...
}

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

protected Map<String,Set<Watcher>> get Watched(int rc){
     return watchManager.dataWatches;
}
public void register(int rc){
      if(shouldAddWatch(rc)){
          Map<String,Set<Watcher>> watches = getWatches(rc);
          synchronized(watched){
               Set<Watcher> watches = watches.get(clientPaht);
                if(watchers== null){
                     watchers = new HashSet<Watcher>();
                      watchers.put(clientPath,watchers);
                 }
                 watchers.add(watcher);
          }
      }
}

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

图7-8 客户端Watcher注册流程图

通过上面的讲解，相信读者已经对客户端的Watcher注册流程有了一个大概的了解。但同时我们也可以发现，极端情况下，客户端美调用一次getData（）接口，就会注册上一个Watcher，那么这些Watcher实体都会随着客户端请求被发送到服务端去吗？

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

public void createBB(){\
     try{
          ByteArrayOutputStream baos = new ByteArrayOutputStream();
           BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
           boa.writeInt(-1,"len");
           if(requestHeader != null){
                 requestHeader.serialize(boa,"header");
           }
           if(request instanceof ConnectRequest){
                 request.serialize(boa,"connect");
                 //append "an-I-allowed-to-be-readonly" flag
                  boa.writeBool(readOnly,"readOnly");
           }else if(request != null){
                request.serialize(boa,"request");
           }
     }
}

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

后面博主会对服务端处理watcher以及客户端的回调进行详细的讲解，这两个章节是ZooKeeper的精髓，希望不了解的进行仔细阅读。