Watcher觸發(fā)

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

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從存儲節(jié)點(diǎn)的ConcurrentHashMap中獲取節(jié)點(diǎn)信息，
  • 修改節(jié)點(diǎn)信息
  • 調(diào)用WatchManager 的triggerWatch方法
• 可以看到以上代碼是通過調(diào)用WatchManager的triggerWatch方法來觸發(fā)相關(guān)事件

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

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方法邏輯比較簡單以下幾個步驟：
  • 將請求頭標(biāo)記為-1，標(biāo)識當(dāng)前是一個通知
  • 將watchedEvent包裝成WatcherEvent，以方便進(jìn)行網(wǎng)絡(luò)傳輸序列化（之前篇解釋過WatcherEvent用來網(wǎng)絡(luò)傳輸用）
  • 向客戶端發(fā)送該通知
• 我們從上面步驟看其實(shí)他并沒有處理客戶端Watcher的邏輯，只是借用當(dāng)前客戶端連接的ServerCnxn對象來實(shí)現(xiàn)對客戶端WatchedEvent的傳遞，真正的客戶端Watcher回調(diào)與業(yè)務(wù)邏輯的執(zhí)行肯定都在客戶端這邊

客戶端回調(diào)Watcher

• 對于一個來自服務(wù)器的通知是通ServerCnxn中發(fā)送出來的，同樣的，客戶端這邊的響應(yīng)也是在一個類似的類中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;}......}

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

EventThread處理事件通知

• 如上流程中，服務(wù)的的Watcher時間通知最終交給了EventThread線程來處理，EventThread是Zookeeper客戶端中專門用來處理服務(wù)器端通知的事件線程，我們看下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方法首先根據(jù)該通知事件從ZKWatchManager中取出所有相關(guān)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后，會從相應(yīng)的Watcher存儲（即上代碼中dataWatches，existWatches，childWatches中一個或者多個，比如NodeCreated 事件類型從dataWatches ,和 existWatches中所有watcher）中去掉對應(yīng)的Watcher，此處用的remove，標(biāo)識客戶端的Watcher機(jī)制統(tǒng)一也是一次性的，觸發(fā)后該Watcher就失效了。
• 接著利用Watcher封裝成一個WatcherSetEventPair，并且將這個對象加入一個阻塞隊(duì)列LinkedBlockingQueue 中，并且我們在ClientCnxn類中能找到一個run方法，這個方法會不斷的從阻塞隊(duì)列中take出數(shù)據(jù)然后再發(fā)送對應(yīng)的通知process進(jìn)行串行同步處理，這里的Watcher才是真正的客戶端注冊的Watcher，調(diào)用這個Watcher的process方法就可以實(shí)現(xiàn)回調(diào)了，保證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特性總結(jié)

• 通過我們上面的分析，了解了Watcher機(jī)制的相關(guān)接口定義以及Watcher的各類事件，我們以Zookeeper節(jié)點(diǎn)的數(shù)據(jù)內(nèi)容過期接口為例，從Zookeeper客戶端進(jìn)行Watcher注冊，服務(wù)的處理Watcher以及客戶端回調(diào)watcher三方面階段講解了ZooKeeper的Watcher工作機(jī)制，發(fā)現(xiàn)Watcher有以下幾個特點(diǎn)：

一次性

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

客戶端串行執(zhí)行

• 客戶端watcher回調(diào)的過程是一個串行同步的過程，這保證了順序性，同事需要開發(fā)人員注意不要因?yàn)橐粋€Watcher的處理邏輯影響了整個客戶端Watcher回調(diào)

輕量級

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

上一篇Zookeeper–Watcher機(jī)制源碼剖析一
下一篇Zookeeper實(shí)踐與應(yīng)用- Canal

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。