Zookeeper源码分析：Watcher机制

1. 设置Watcher 使用Watcher需要先实现Watcher接口，并将实现类对象传递到指定方法中，如getChildren, exist等。Zookeeper允许在构造Zookeeper对象时候指定一个默认Watcher对象.getChildren和exit方法可以使用这个默认的Watcher对象，也可以指定一个新Watcher对象。

Code 1: Watcher接口

public interface Watcher {

    /**     * Event的状态     */     public interface Event {         /**         * 在事件发生时，ZooKeeper的状态         */         public enum KeeperState {

            @Deprecated             Unknown (-1),

            Disconnected (0),

            @Deprecated             NoSyncConnected (1),

            SyncConnected (3),

            AuthFailed (4),

            ConnectedReadOnly (5),

            SaslAuthenticated(6),

            Expired (-112);

            private final int intValue; 

            KeeperState( int intValue) {                 this.intValue = intValue;             } 

            ......         }

        /**         * ZooKeeper中的事件         */         public enum EventType {             None (-1),             NodeCreated (1),             NodeDeleted (2),             NodeDataChanged (3),             NodeChildrenChanged (4);

            private final int intValue;    // Integer representation of value                                             // for sending over wire             EventType( int intValue) {                 this.intValue = intValue;             }             ......          }     }

    //Watcher的回调方法     abstract public void process(WatchedEvent event); }

Code 2: Zookeeper.getChildren（final String, Watcher)方法

public List<String> getChildren(final String path, Watcher watcher)     throws KeeperException, InterruptedException {     final String clientPath = path;     PathUtils. validatePath(clientPath);

    WatchRegistration wcb = null;     //如果watcher不等于null, 构建WatchRegistration对象，     //该对象描述了watcher和path之间的关系     if (watcher != null) {         wcb = new ChildWatchRegistration(watcher, clientPath);     }     //在传入的path加上root path前缀，构成服务器端的绝对路径     final String serverPath = prependChroot(clientPath);     //构建RequestHeader对象     RequestHeader h = new RequestHeader();     //设置操作类型为OpCode. getChildren     h.setType(ZooDefs.OpCode. getChildren);     //构建GetChildrenRequest对象     GetChildrenRequest request = new GetChildrenRequest();     //设置path     request.setPath(serverPath);     //设置是否使用watcher     request.setWatch(watcher != null);     //构建GetChildrenResponse对象     GetChildrenResponse response = new GetChildrenResponse();     //提交请求，并阻塞等待结果     ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);     if (r.getErr() != 0) {         throw KeeperException.create(KeeperException.Code. get(r.getErr()),                 clientPath);     }     return response.getChildren(); }

Follower的NIOServerCnxn类接到了Client的请求，会调用ZookeeperServer.processPacket()方法。该方法会构建一个Request对象，并调用第一个处理器FollowerRequestProcessor。

由于我们的请求只是一个读操作，而不是一个Quorum请求或者sync请求，所以FollowerRequestProcessor不需要调用Follower.request()方法将请求转给Leader，只需要将请求传递到下一个处理器CommitProcessor。

处理器CommitProcessor线程发现请求是读请求后，直接将Requet对象加入到toProcess队列中，在接下的循环中会调用FinalRequestProcessor.processRequest方法进行处理。

FinalRequestProcessor.processRequest方法最终会调用ZKDatabase中的读操作方法（如statNode和getData方法）， 而ZKDatabase的这些方法会最终调用DataTree类的方法来获取指定path的znode信息并返回给Client端，同时也会设置Watcher。

Code 3: FinalRequestProcessor对OpCode.getData请求的处理

case OpCode. getData: {               lastOp = "GETD";               GetDataRequest getDataRequest = new GetDataRequest();               ByteBufferInputStream. byteBuffer2Record(request.request,                       getDataRequest);               //获得znode对象               DataNode n = zks.getZKDatabase().getNode(getDataRequest.getPath());               //n为null, 抛出NoNodeException异常               if (n == null) {                   throw new KeeperException.NoNodeException();               }               Long aclL;               synchronized(n) {                   aclL = n. acl;               }               //检查是否有读权限               PrepRequestProcessor. checkACL(zks, zks.getZKDatabase().convertLong(aclL),                       ZooDefs.Perms. READ,                       request. authInfo);               //构建状态对象stat               Stat stat = new Stat();               //获得指定path的znode数据，               //如果GetDataRequest.getWatcher()返回true, 将ServerCnxn类型对象cnxn传递进去。               //ServerCnxn是实现了Watcher接口               byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat,                       getDataRequest. getWatch() ? cnxn : null);               //构建GetDataResponse对象               rsp = new GetDataResponse(b, stat);               break;           }

Code 4: DataTree.getData()方法

public byte[] getData(String path, Stat stat, Watcher watcher)         throws KeeperException.NoNodeException {     //从nodes map中获取指定path的DataNode对象     DataNode n = nodes.get(path);     //如果n为null, 则抛出NoNodeException异常     if (n == null) {         throw new KeeperException.NoNodeException();     }     synchronized (n) {         //将n的状态copy到stat中         n.copyStat(stat);         //如果watcher不会null, 则将(path, watcher)键值对放入dataWatchers Map里         if (watcher != null) {             dataWatches.addWatch(path, watcher);         }         //返回节点数据         return n.data ;     } }

2. 修改znode数据触发Watcher 在Zookeeper二阶段提交的COMMIT阶段。当Follower从Leader那接收到一个写请求的Leader.COMMIT数据包，会调用FinalRequestProcessor.processRequest()方法。Leader本身在发送完Leader.COMMIT数据包，也会调用FinalRequestProcessor.processRequest()方法。

如果是setData修改数据请求，那么FinalRequestProcessor.processRequest()方法最终会调用到DataTree.setData方法将txn应用到指定znode上，同时触发Watcher，并发送notification给Client端。

其关SetData请求的时序图如下：

triggerWatcher

Code 5: DataTree.setData()方法

public Stat setData(String path, byte data[], int version, long zxid,         long time) throws KeeperException.NoNodeException {     Stat s = new Stat();     //根据path, 获得DataNode对象n     DataNode n = nodes.get(path);     //如果n为null, 则抛出NoNodeException异常     if (n == null) {         throw new KeeperException.NoNodeException();     }     byte lastdata[] = null;     synchronized (n) {         lastdata = n. data;         n. data = data;         n. stat.setMtime(time);         n. stat.setMzxid(zxid);         n. stat.setVersion(version);         n.copyStat(s);     }     // now update if the path is in a quota subtree.     String lastPrefix = getMaxPrefixWithQuota(path);     if(lastPrefix != null) {       this.updateBytes(lastPrefix, (data == null ? 0 : data.length)           - (lastdata == null ? 0 : lastdata.length ));     }     //触发Watcher     dataWatches.triggerWatch(path, EventType.NodeDataChanged);     return s; }

Code 6: WatchManage.triggerWatcher()方法，触发Watcher。

Set<Watcher> triggerWatch(String path, EventType type, Set<Watcher> supress) {     WatchedEvent e = new WatchedEvent(type,             KeeperState. SyncConnected, path);     HashSet<Watcher> watchers;     synchronized (this ) {         //从watchTable删除掉path对于的watcher         watchers = watchTable.remove(path);         if (watchers == null || watchers.isEmpty()) {             if (LOG .isTraceEnabled()) {                 ZooTrace. logTraceMessage(LOG,                         ZooTrace. EVENT_DELIVERY_TRACE_MASK,                         "No watchers for " + path);             }             return null;         }         for (Watcher w : watchers) {             HashSet<String> paths = watch2Paths.get(w);             if (paths != null) {                 paths.remove(path);             }         }     }     //循环处理所有关于path的Watcher, 这里Watcher对象实际上就是ServerCnxn类型对象     for (Watcher w : watchers) {         if (supress != null && supress.contains(w)) {             continue;         }         w.process(e);     }     return watchers; }

Code 7: NIOServerCnxn.process方法，发送notification给Client端

synchronized public void process (WatchedEvent event) {     ReplyHeader h = new ReplyHeader(-1, -1L, 0);     if (LOG .isTraceEnabled()) {         ZooTrace. logTraceMessage(LOG, ZooTrace.EVENT_DELIVERY_TRACE_MASK ,                                 "Deliver event " + event + " to 0x"                                 + Long. toHexString(this. sessionId)                                 + " through " + this );     }

    // Convert WatchedEvent to a type that can be sent over the wire     WatcherEvent e = event.getWrapper();     //发送notification给Client端     sendResponse(h, e, "notification"); }

3. 总结 Watcher具有one-time trigger的特性，在代码中我们也可以看到一个watcher被处理后会立即从watchTable中删掉。