Zookeeper 通知更新可靠嗎？ 解讀原始碼找答案！
歡迎大家前往騰訊雲+社群，獲取更多騰訊海量技術實踐乾貨哦~
導讀：
遇到Keepper通知更新無法收到的問題，思考節點變更通知的可靠性，通過閱讀原始碼解析瞭解到zk Watch的註冊以及觸發的機制，本地除錯執行模擬zk更新的不可靠的場景以及得出相應的解決方案。
過程很曲折，但問題的根本原因也水落石出了，本文最後陳述了更新無法收到的根本原因，希望對其他人有所幫助。-----------------------------------------
通常Zookeeper是作為配置儲存、分散式鎖等功能被使用，配置讀取如果每一次都是去Zookeeper server讀取效率是非常低的，幸好Zookeeper提供節點更新的通知機制，只需要對節點設定Watch監聽，節點的任何更新都會以通知的方式傳送到Client端。
如上圖所示：應用Client通常會連線上某個ZkServer，forPath不僅僅會讀取Zk 節點zkNode的資料（通常儲存讀取到的資料會儲存在應用記憶體中，例如圖中Value），而且會設定一個Watch，當zkNode節點有任何更新時，ZkServer會發送notify，Client執行Watch來才走出相應的事件相應。這裡假設操作為更新Client本地的資料。這樣的模型使得配置非同步更新到Client中，而無需Client每次都遠端讀取，大大提高了讀的效能，（圖中的re-regist重新註冊是因為對節點的監聽是一次性的，每一次通知完後，需要重新註冊）。但這個Notify是可靠的嗎？如果通知失敗，那豈不是Client永遠都讀取的本地的未更新的值？
由於現網環境定位此類問題比較困難，因此本地下載原始碼並模擬執行ZkServer & ZkClient來看通知的傳送情況。
2、cd 到路徑下，執行ant eclipse 載入工程的依賴。
3、匯入Idea中。
檢視相關問題和步驟。
首先執行ZkServer。QuorumPeerMain是Server的啟動類。這個可以根據bin下ZkServer.sh找到入口。注意啟動引數配置引數檔案，指定例如啟動埠等相關引數。
在此之前，需要設定相關的斷點。
首先我們要看client設定監聽後，server是如何處理的
ZkClient 是使用Nio的方式與ZkServer進行通訊的，Zookeeper的執行緒模型中使用兩個執行緒：
SendThread專門成立的請求的傳送，請求會被封裝為Packet（包含節點名稱、Watch描述等資訊）類傳送給Sever。
EventThread則專門處理SendThread接收後解析出的Event。
ZkClient 的主要有兩個Processor，一個是SycProcessor負責Cluster之間的資料同步（包括叢集leader選取）。另一個是叫FinalRuestProcessor，專門處理對接受到的請求（Packet）進行處理。
//ZookeeperServer 的processPacket方法專門對收到的請求進行處理。
public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException {
// We have the request, now process and setup for next
InputStream bais = new ByteBufferInputStream(incomingBuffer);
BinaryInputArchive bia = BinaryInputArchive.getArchive(bais);
RequestHeader h = new RequestHeader();
h.deserialize(bia, "header");
// Through the magic of byte buffers, txn will not be
// pointing
// to the start of the txn
incomingBuffer = incomingBuffer.slice();
//鑑權請求處理
if (h.getType() == OpCode.auth) {
LOG.info("got auth packet " + cnxn.getRemoteSocketAddress());
AuthPacket authPacket = new AuthPacket();
ByteBufferInputStream.byteBuffer2Record(incomingBuffer, authPacket);
String scheme = authPacket.getScheme();
ServerAuthenticationProvider ap = ProviderRegistry.getServerProvider(scheme);
Code authReturn = KeeperException.Code.AUTHFAILED;
if(ap != null) {
try {
authReturn = ap.handleAuthentication(new ServerAuthenticationProvider.ServerObjs(this, cnxn), authPacket.getAuth());
} catch(RuntimeException e) {
LOG.warn("Caught runtime exception from AuthenticationProvider: " + scheme + " due to " + e);
authReturn = KeeperException.Code.AUTHFAILED;
}
}
if (authReturn == KeeperException.Code.OK) {
if (LOG.isDebugEnabled()) {
LOG.debug("Authentication succeeded for scheme: " + scheme);
}
LOG.info("auth success " + cnxn.getRemoteSocketAddress());
ReplyHeader rh = new ReplyHeader(h.getXid(), 0,
KeeperException.Code.OK.intValue());
cnxn.sendResponse(rh, null, null);
} else {
if (ap == null) {
LOG.warn("No authentication provider for scheme: "
+ scheme + " has "
+ ProviderRegistry.listProviders());
} else {
LOG.warn("Authentication failed for scheme: " + scheme);
}
// send a response...
ReplyHeader rh = new ReplyHeader(h.getXid(), 0,
KeeperException.Code.AUTHFAILED.intValue());
cnxn.sendResponse(rh, null, null);
// ... and close connection
cnxn.sendBuffer(ServerCnxnFactory.closeConn);
cnxn.disableRecv();
}
return;
} else {
if (h.getType() == OpCode.sasl) {
Record rsp = processSasl(incomingBuffer,cnxn);
ReplyHeader rh = new ReplyHeader(h.getXid(), 0, KeeperException.Code.OK.intValue());
cnxn.sendResponse(rh,rsp, "response"); // not sure about 3rd arg..what is it?
return;
}
else {
Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(),
h.getType(), incomingBuffer, cnxn.getAuthInfo());
si.setOwner(ServerCnxn.me);
// Always treat packet from the client as a possible
// local request.
setLocalSessionFlag(si);
//交給finalRequestProcessor處理
submitRequest(si);
}
}
cnxn.incrOutstandingRequests(h);
}FinalRequestProcessor 對請求進行解析，Client連線成功後，傳送的exist命令會落在這部分處理邏輯。
zkDataBase 由zkServer從disk持久化的資料建立而來，上圖可以看到這裡就是新增監聽Watch的地方。
然後我們需要了解到，當Server收到節點更新事件後，是如何觸發Watch的。
首先了解兩個概念，FinalRequestProcessor處理的請求分為兩種，一種是事務型的，一種非事務型，exist 的event-type是一個非事物型的操作，上面程式碼中是對其處理邏輯，對於事物的操作，例如SetData的操作。則在下面程式碼中處理。
private ProcessTxnResult processTxn(Request request, TxnHeader hdr,
Record txn) {
ProcessTxnResult rc;
int opCode = request != null ? request.type : hdr.getType();
long sessionId = request != null ? request.sessionId : hdr.getClientId();
if (hdr != null) {
//hdr 為事物頭描述，例如SetData的操作就會被ZkDataBase接管操作，
//因為是對Zk的資料儲存機型修改
rc = getZKDatabase().processTxn(hdr, txn);
} else {
rc = new ProcessTxnResult();
}
if (opCode == OpCode.createSession) {
if (hdr != null && txn instanceof CreateSessionTxn) {
CreateSessionTxn cst = (CreateSessionTxn) txn;
sessionTracker.addGlobalSession(sessionId, cst.getTimeOut());
} else if (request != null && request.isLocalSession()) {
request.request.rewind();
int timeout = request.request.getInt();
request.request.rewind();
sessionTracker.addSession(request.sessionId, timeout);
} else {
LOG.warn("*****>>>>> Got "
+ txn.getClass() + " "
+ txn.toString());
}
} else if (opCode == OpCode.closeSession) {
sessionTracker.removeSession(sessionId);
}
return rc;
}
這裡設定了斷點，就可以攔截對節點的更新操作。
這兩個設定了斷點，就可以瞭解到Watch的設定過程。
接下來看如何啟動Zookeeper的Client。ZookeeperMain為Client的入口，同樣在bin/zkCli.sh中可以找到。注意設定引數，設定Server的連線地址。
修改ZookeeperMain方法，設定對節點的Watch監聽。
public ZooKeeperMain(String args[]) throws IOException, InterruptedException, KeeperException {
cl.parseOptions(args);
System.out.println("Connecting to " + cl.getOption("server"));
connectToZK(cl.getOption("server"));
while (true) {
// 模擬註冊對/zookeeper節點的watch監聽
zk.exists("/zookeeper", true);
System.out.println("wait");
}
}啟動Client。
由於我們要觀察節點變更的過程，上面這個Client設定了對節點的監聽，那麼我們需要另外一個cleint對節點進行更改，這個我們只需要在命令上進行就可以了。
此時命令列的zkClient更新了/zookeeper節點，Server此時會停在setData事件的處理程式碼段。
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 = getMaxPrefixWithQuota(path);
if(lastPrefix != null) {
this.updateBytes(lastPrefix, (data == null ? 0 : data.length)
- (lastdata == null ? 0 : lastdata.length));
}
//觸發watch監聽
dataWatches.triggerWatch(path, EventType.NodeDataChanged);
return s;
}此時，我們重點關注的類出現了。WatchManager
package org.apache.zookeeper.server;
import java.io.PrintWriter;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* This class manages watches. It allows watches to be associated with a string
* and removes watchers and their watches in addition to managing triggers.
*/
class WatchManager {
private static final Logger LOG = LoggerFactory.getLogger(WatchManager.class);
//儲存path對watch的關係
private final Map<String, Set<Watcher>> watchTable =
new HashMap<String, Set<Watcher>>();
//儲存watch監聽了哪些path節點
private final Map<Watcher, Set<String>> watch2Paths =
new HashMap<Watcher, Set<String>>();
synchronized int size(){
int result = 0;
for(Set<Watcher> watches : watchTable.values()) {
result += watches.size();
}
return result;
}
//新增監聽
synchronized void addWatch(String path, Watcher watcher) {
Set<Watcher> list = watchTable.get(path);
if (list == null) {
// don't waste memory if there are few watches on a node
// rehash when the 4th entry is added, doubling size thereafter
// seems like a good compromise
list = new HashSet<Watcher>(4);
watchTable.put(path, list);
}
list.add(watcher);
Set<String> paths = watch2Paths.get(watcher);
if (paths == null) {
// cnxns typically have many watches, so use default cap here
paths = new HashSet<String>();
watch2Paths.put(watcher, paths);
}
paths.add(path);
}
//移除
synchronized void removeWatcher(Watcher watcher) {
Set<String> paths = watch2Paths.remove(watcher);
if (paths == null) {
return;
}
for (String p : paths) {
Set<Watcher> list = watchTable.get(p);
if (list != null) {
list.remove(watcher);
if (list.size() == 0) {
watchTable.remove(p);
}
}
}
}
Set<Watcher> triggerWatch(String path, EventType type) {
return triggerWatch(path, type, null);
}
//觸發watch
Set<Watcher> triggerWatch(String path, EventType type, Set<Watcher> supress) {
WatchedEvent e = new WatchedEvent(type,
KeeperState.SyncConnected, path);
Set<Watcher> watchers;
synchronized (this) {
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) {
Set<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的方法，可以發現watch被移除後，即往watch中儲存的client資訊進行通知傳送。
@Override
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();
sendResponse(h, e, "notification");
}沒有任何確認機制，不會由於傳送失敗，而回寫watch。
結論:
到這裡，可以知道watch的通知機制是不可靠的，zkServer不會保證通知的可靠抵達。雖然zkclient與zkServer端是會有心跳機制保持連結，但是如果通知過程中斷開，即時重新建立連線後，watch的狀態是不會恢復。
現在已經知道了通知是不可靠的，會有丟失的情況，那ZkClient的使用需要進行修正。
本地的儲存不再是一個靜態的等待watch更新的狀態，而是引入快取機制，定期的去從Zk主動拉取並註冊Watch（ZkServer會進行去重，對同一個Node節點的相同時間型別的Watch不會重複）。
另外一種方式是，Client端收到斷開連線的通知，重新註冊所有關注節點的Watch。但作者遇到的現網情況是client沒有收到更新通知的同時，也沒有檢視到連線斷開的錯誤資訊。這塊仍需進一步確認。水平有限，歡迎指正 :D
在StackOverFlow上的提問有了新進展:
原來官方文件已經解釋了在連線斷開的時候，client對watch的一些恢復操做，ps：原來上面我提到的客戶端的策略已經官方實現。。。
客戶端會通過心跳保活，如果發現斷開了連線，會重新建立連線，併發送之前對節點設定的watch以及節點zxid，如果zxid與服務端的小則說明斷開期間有更改，那麼server會觸發通知。
這麼來看，Zookeeper的通知機制至少在官方的文件說明上是可靠的，至少是有相應機制去保證。ps：除Exist watch外。但是本人遇到的問題仍未解開。。後悔當初沒有保留現場，深入發掘。計劃先把實現改回原來的，後續進一步驗證。找到原因再更新這裡。
最終結論更新！
通過深入閱讀apache的zk論壇以及原始碼，有一個重要的資訊。
上面提到的連線斷開分為recoverble以及unrecoverble兩種場景，這兩種的區別主要是基於Session的有效期，所有的client操作包括watch都是和Session關聯的，當Session在超時過期時間內，重新成功建立連線，則watch會在連線建立後重新設定。但是當Session Timeout後仍然沒有成功重新建立連線，那麼Session則處於Expire的狀態。下面連線講述了這個過程
這種情況下，ZookeeperClient會重新連線，但是Session將會是全新的一個。同時之前的狀態是不會儲存的。
private void conLossPacket(Packet p) {
if (p.replyHeader == null) {
return;
}
switch (state) {
case AUTH_FAILED:
p.replyHeader.setErr(KeeperException.Code.AUTHFAILED.intValue());
break;
case CLOSED:
// session關閉狀態，直接返回。
p.replyHeader.setErr(KeeperException.Code.SESSIONEXPIRED.intValue());
break;
default:
p.replyHeader.setErr(KeeperException.Code.CONNECTIONLOSS.intValue());
}
// 如果session未過期，這裡進行session的狀態（watches）會重新註冊。
finishPacket(p);
}*1、什麼是zookeeper的會話過期？*
一般來說，我們使用zookeeper是叢集形式，如下圖，client和zookeeper叢集(3個例項)建立一個會話session。
在這個會話session當中，client其實是隨機與其中一個zk provider建立的連結，並且互發心跳heartbeat。zk叢集負責管理這個session，並且在所有的provider上維護這個session的資訊，包括這個session中定義的臨時資料和監視點watcher。
如果再網路不佳或者zk叢集中某一臺provider掛掉的情況下，有可能出現connection loss的情況，例如client和zk provider1連線斷開，這時候client不需要任何的操作(zookeeper api已經給我們做好了)，只需要等待client與其他provider重新連線即可。這個過程可能導致兩個結果：
1）在session timeout之內連線成功
這個時候client成功切換到連線另一個provider例如是provider2，由於zk在所有的provider上同步了session相關的資料，此時可以認為無縫遷移了。
2）在session timeout之內沒有重新連線
這就是session expire的情況，這時候zookeeper叢集會任務會話已經結束，並清除和這個session有關的所有資料，包括臨時節點和註冊的監視點Watcher。
在session超時之後，如果client重新連線上了zookeeper叢集，很不幸，zookeeper會發出session expired異常，且不會重建session，也就是不會重建臨時資料和watcher。
我們實現的ZookeeperProcessor是基於Apache Curator的Client封裝實現的。
它對於Session Expire的處理是提供了處理的監聽註冊ConnectionStateListner，當遇到Session Expire時，執行使用者要做的邏輯。（例如：重新設定Watch）遺憾的是，我們沒有對這個事件進行處理，因此連線是一致斷開的，但是！我們應用仍然會讀到老的資料！
在這裡，我們又犯了另外一個錯誤，本地快取了zookeeper的節點資料。。其實zookeeperClient已經做了本地快取的機制，但是我們有加了一層（注：這裡也有一個原因，是因為zk節點的資料時二進位制的陣列，業務要使用通常要反序列化，我們這裡的快取是為了減少反序列化帶來的開銷！），正式由於我們本地快取了，因此即使zk斷開了，仍然讀取了老的值！
至此，謎團已經全部解開，看來之前的實現有許多姿勢是錯誤的，導致後續出現了各種奇怪的BUG 。現在處理的方案，是監聽Reconnect的通知，當收到這個通知後，主動讓本地快取失效（這裡仍然做了快取，是因為減少反序列化的開銷，zkClient的快取只是快取了二進位制，每次拿出來仍然需要反序列化）。程式碼：
curatorFramework.getConnectionStateListenable().addListener(new ConnectionStateListener() {
@Override
public void stateChanged(CuratorFramework client, ConnectionState newState) {
switch (newState) {
case CONNECTED:
break;
case RECONNECTED:
LOG.error("zookeeper connection reconnected");
System.out.println("zookeeper connection reconnected");
//本來使用invalidateAll，但是這個會使得cache所有快取值同時失效
//如果關注節點比較多，導致同時請求zk讀值，可能服務會瞬時阻塞在這一步
//因此使用guava cache refresh方法，非同步更新，更新過程中，
//老值返回，知道更新完成
for (String key : classInfoMap.keySet()) {
zkDataCache.refresh(key);
}
break;
case LOST:
// session 超時，斷開連線，這裡不要做任何操作，快取保持使用
LOG.error("zookeeper connection lost");
System.out.println("zookeeper connection lost");
break;
case SUSPENDED:
break;
default:
break;
}
}
});此文已由作者授權騰訊雲+社群釋出，更多原文請點選
搜尋關注公眾號「雲加社群」，第一時間獲取技術乾貨，關注後回覆1024 送你一份技術課程大禮包！
海量技術實踐經驗，盡在雲加社群！