Watcher觸發

• 我們從實際操作時候的表現來看Watcher的觸發，比如Zookeeper中NodeDataChanged時間的觸發是“Watcher監聽的對應數據節點的數據內容發生變更”，需要修改節點數據那么必然和數據節點存儲的位置DataTree有關系，我們從這里去尋找修改后觸發Watcher的答案。
• 我們從DataTree類中找到了修改節點的入口setData方法，我們從上篇內容中也知道了ServerCnxn存儲到WatchManager中，并且以不同的維度存儲了兩份數據：

public Stat setData(String path, byte data[], int version, long zxid,long time) throws KeeperException.NoNodeException {Stat s = new Stat();DataNode n = nodes.get(path);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;if((lastPrefix = getMaxPrefixWithQuota(path)) != null) {this.updateBytes(lastPrefix, (data == null ? 0 : data.length)- (lastdata == null ? 0 : lastdata.length));}dataWatches.triggerWatch(path, EventType.NodeDataChanged);return s;}

• 以上setData方法流程就兩個步驟：
  • 利用Path從存儲節點的ConcurrentHashMap中獲取節點信息，
  • 修改節點信息
  • 調用WatchManager 的triggerWatch方法
• 可以看到以上代碼是通過調用WatchManager的triggerWatch方法來觸發相關事件

public Set<Watcher> triggerWatch(String path, EventType type, Set<Watcher> supress) {WatchedEvent e = new WatchedEvent(type,KeeperState.SyncConnected, path);HashSet<Watcher> watchers;synchronized (this) {watchers = watchTable.remove(path);......for (Watcher w : watchers) {HashSet<String> paths = watch2Paths.get(w);if (paths != null) {paths.remove(path);}}}for (Watcher w : watchers) {if (supress != null && supress.contains(w)) {continue;}w.process(e);}return watchers;}

• 如上是triggerWatch源碼中的觸發邏輯有如下幾個步驟：
  • 封裝WatchedEvent：首先從通知參數中獲取到通知狀態KeeperState，事件類型EventType，節點路徑Path封裝成一個WatchedEvent對象
  • 查詢Watcher對象：根據數據節點的節點路徑從WatchTable中取出對應的Watcher，如果沒有找到watcher，說明沒有任何客戶端在這個節點上注冊過Watcher，直接退出。找到了這個Watcher將他取出來，同時直接從watchTable和watch2Path中刪掉------這個步驟可以看出，watcher在服務端是一次性的，即觸發一次就失效了。
  • 調用process方法來觸發Watcher： 在最后的for循環中依次取出每一個watcher來調用每一個的process，我們看一下他的實現類有N多個如下圖，實際調用是哪一個呢，我們得從之前的注冊代碼中去找答案。之前Zookeeper服務器注冊到WatchManager中的是watcher的實現類ServerCnxn，所有我們直接看ServerCnxn的process實現方法就可以
    
• 以下ServerCnxn對應的process是一個抽象方法，他的實現是NIOServerCnxn的實現：

public abstract void process(WatchedEvent event); 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 wireWatcherEvent e = event.getWrapper();sendResponse(h, e, "notification");}

• 以上代碼片段可以看出process方法邏輯比較簡單以下幾個步驟：
  • 將請求頭標記為-1，標識當前是一個通知
  • 將watchedEvent包裝成WatcherEvent，以方便進行網絡傳輸序列化（之前篇解釋過WatcherEvent用來網絡傳輸用）
  • 向客戶端發送該通知
• 我們從上面步驟看其實他并沒有處理客戶端Watcher的邏輯，只是借用當前客戶端連接的ServerCnxn對象來實現對客戶端WatchedEvent的傳遞，真正的客戶端Watcher回調與業務邏輯的執行肯定都在客戶端這邊

客戶端回調Watcher

• 對于一個來自服務器的通知是通ServerCnxn中發送出來的，同樣的，客戶端這邊的響應也是在一個類似的類中ClientCnxn中，ClientCnxn中通過SendThread來收事件通知

SendThread接收事件通知

• 我們來看下ClientCnxn的接收通知的源碼處理：

class SendThread extends ZooKeeperThread {private long lastPingSentNs;private final ClientCnxnSocket clientCnxnSocket;private Random r = new Random(System.nanoTime()); private boolean isFirstConnect = true;void readResponse(ByteBuffer incomingBuffer) throws IOException {ByteBufferInputStream bbis = new ByteBufferInputStream(incomingBuffer);BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);ReplyHeader replyHdr = new ReplyHeader();replyHdr.deserialize(bbia, "header");......if (replyHdr.getXid() == -1) {// -1 means notificationif (LOG.isDebugEnabled()) {LOG.debug("Got notification sessionid:0x"+ Long.toHexString(sessionId));}WatcherEvent event = new WatcherEvent();event.deserialize(bbia, "response");// convert from a server path to a client pathif (chrootPath != null) {String serverPath = event.getPath();if(serverPath.compareTo(chrootPath)==0)event.setPath("/");else if (serverPath.length() > chrootPath.length())event.setPath(serverPath.substring(chrootPath.length()));else {LOG.warn("Got server path " + event.getPath()+ " which is too short for chroot path "+ chrootPath);}}WatchedEvent we = new WatchedEvent(event);if (LOG.isDebugEnabled()) {LOG.debug("Got " + we + " for sessionid 0x"+ Long.toHexString(sessionId));}eventThread.queueEvent( we );return;}......}

• 對于一個服務端的響應客戶端由SendThread.readResponse(ByteBuffer incomingBuffer)方法來統一處理，如果響應頭replyHdr中標識的XID為-1，標識這個是一個通知類型響應，對其的處理大體上分為4個步驟
  • 反序列化：replyHdr.deserialize(bbia, “header”); 方法Zookeeper客戶端接收到請求首先將字節流轉換成WatcherEvent對象
  • 處理Chrootpath：如果客戶端設置了chrootPath屬性，那么要對服務端傳過來的完整的節點路徑進行chrootPath處理，生成客戶端的一個相對節點路徑，例如客戶端設置差rootPath為/appl，那么針對服務器端傳來的響應節點路徑為/appl/locks，經過chrootPath處理后，就變成相對路徑/locks。
  • 還原WatchedEvent：通過接收到的WatcherEvent得到WatchedEvent
  • 回調Watcher，最后將WatchedEvent對象交給EventThread線程，在下一個輪詢周期中進行Watcher回調。

EventThread處理事件通知

• 如上流程中，服務的的Watcher時間通知最終交給了EventThread線程來處理，EventThread是Zookeeper客戶端中專門用來處理服務器端通知的事件線程，我們看下EventThread中是怎么處理的。由上面代碼中queueEvent 方法入口：

public void queueEvent(WatchedEvent event) {if (event.getType() == EventType.None&& sessionState == event.getState()) {return;}sessionState = event.getState();// materialize the watchers based on the eventWatcherSetEventPair pair = new WatcherSetEventPair(watcher.materialize(event.getState(), event.getType(),event.getPath()),event);// queue the pair (watch set & event) for later processingwaitingEvents.add(pair);}

• 如上 QueueEvent方法首先根據該通知事件從ZKWatchManager中取出所有相關Watcher，materalize方法如下：

private final Map<String, Set<Watcher>> dataWatches =new HashMap<String, Set<Watcher>>();private final Map<String, Set<Watcher>> existWatches =new HashMap<String, Set<Watcher>>();private final Map<String, Set<Watcher>> childWatches =new HashMap<String, Set<Watcher>>();public Set<Watcher> materialize(Watcher.Event.KeeperState state,Watcher.Event.EventType type,String clientPath){Set<Watcher> result = new HashSet<Watcher>();switch (type) {case None:......case NodeDataChanged:case NodeCreated:synchronized (dataWatches) {addTo(dataWatches.remove(clientPath), result);}synchronized (existWatches) {addTo(existWatches.remove(clientPath), result);}break;......}}final private void addTo(Set<Watcher> from, Set<Watcher> to) {if (from != null) {to.addAll(from);}}

• 以上代碼中客戶端識別出EventType后，會從相應的Watcher存儲（即上代碼中dataWatches，existWatches，childWatches中一個或者多個，比如NodeCreated 事件類型從dataWatches ,和 existWatches中所有watcher）中去掉對應的Watcher，此處用的remove，標識客戶端的Watcher機制統一也是一次性的，觸發后該Watcher就失效了。
• 接著利用Watcher封裝成一個WatcherSetEventPair，并且將這個對象加入一個阻塞隊列LinkedBlockingQueue 中，并且我們在ClientCnxn類中能找到一個run方法，這個方法會不斷的從阻塞隊列中take出數據然后再發送對應的通知process進行串行同步處理，這里的Watcher才是真正的客戶端注冊的Watcher，調用這個Watcher的process方法就可以實現回調了，保證FIFO。

@Overridepublic void run() {try {isRunning = true;while (true) {Object event = waitingEvents.take();if (event == eventOfDeath) {wasKilled = true;} else {processEvent(event);}......}} catch (InterruptedException e) { ......} ......}

Watcher特性總結

• 通過我們上面的分析，了解了Watcher機制的相關接口定義以及Watcher的各類事件，我們以Zookeeper節點的數據內容過期接口為例，從Zookeeper客戶端進行Watcher注冊，服務的處理Watcher以及客戶端回調watcher三方面階段講解了ZooKeeper的Watcher工作機制，發現Watcher有以下幾個特點：

一次性

• 無論是客戶端還是服務端，一個Watcher被觸發，ZooKeeper都會將其從相應的存儲中移除，因此，開發人員在Watcher的使用上要記住的一點是要反復注冊，這樣的設計有效的減輕了服務端的壓力，如果注冊一個Watcher后一直有效，針對那些非常頻繁的節點，服務端會不斷向客戶端發送事件通知，無論對網絡還是服務器性能都有非常大影響

客戶端串行執行

• 客戶端watcher回調的過程是一個串行同步的過程，這保證了順序性，同事需要開發人員注意不要因為一個Watcher的處理邏輯影響了整個客戶端Watcher回調

輕量級

• WatchedEvent是整個ZooKeeperWatcher通知機制的最小單元，這個數據結構只包含三部分內容：通知狀態，事件類型，節點路徑。也就是說，Watcher通知非常簡單，只告訴客戶端發生了事件，而不說明具體的內容。
  • 例如針對NodeDataChanged時間ZooKeeper的Watcher只通知客戶端指定數據節點的數據內容發生了變更，而對于原始數據以及變更后的數據都無法直接獲取，需要主動獲取，這也是Watcher機制的一個重要特性
• 另外客戶端向服務端注冊Watcher時候，并不會吧客戶端真實的watcher對象傳到服務端，僅僅是在客戶端請求中使用boolean類型屬性標記，同事服務端也只保存當前連接的ServerCnxn對象
• 輕量級設計使Watcher機制在網絡開銷和服務端內存開銷上都非常廉價。

上一篇Zookeeper–Watcher機制源碼剖析一
下一篇Zookeeper實踐與應用- Canal

總結

以上是生活随笔為你收集整理的Zookeeper--Watcher机制源码剖析二的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。