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請求的時序圖如下：

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中刪掉。

本部落格系博主原創，轉載請附上原部落格地址：http://blog.csdn.net/jeff_fangji/article/details/43910113