深刻淺出Zookeeper（四）：Watch實現剖析
本文首發於 泊浮目的簡書: https://www.jianshu.com/u/204...
| 版本 | 日期 | 備註 |
|---|---|---|
| 1.0 | 2020.4.8 | 文章首發 |
前言
用過Zookeeper的同窗都知道watch是一個很是好用的機制，今天咱們就來看看它的實現原理。java
在正文開始前，咱們先來簡單回憶一下watch是啥？node
Zk提供了分佈式數據的發佈/訂閱功能——這種模型很是的常見，其定義了一種一對多的訂閱關係，可以讓多個訂閱者同時監聽某個主題對象，當這個主題對象自身狀態變化時，則會通知全部訂閱者。具體來講，則是Zk容許一個客戶端向服務端註冊一個watch監聽，當服務端的一些指定事件觸發了這個watch，那麼就會向該客戶端發送事件通知。web
實現
在剖許其實現前，咱們不妨來想想，若是本身動手實現一個watch機制，該怎麼作呢？apache
最簡單的方法是在client保存當前節點的版本，並去輪詢這個節點的狀態。若是發現版本變化，則client觸發watch。不過比起輪詢，不是有更多的好方法，不是嗎？promise
輪詢會給服務器帶來不小的壓力，或許咱們能夠考慮採用相似webhook的方式，讓server保存和client約定好的地址，當watch的數據節點發生變化時，便通知client。服務器
想到這兒，其實已經和Zk本身的watch實現有點像了。沿着這個思路，咱們開始剖析：網絡
通常咱們在使用client時，getData、getChildren、exist均可以用來向Zk註冊watcher。其原理都是同樣的，咱們以exist方法爲例子進行剖析：session
客戶端實現
Zookeeper.exists
/**
* The asynchronous version of exists.
*
* @see #exists(String, Watcher)
*/
public void exists(final String path, Watcher watcher,
StatCallback cb, Object ctx)
{
final String clientPath = path;
PathUtils.validatePath(clientPath);
// the watch contains the un-chroot path
WatchRegistration wcb = null;
if (watcher != null) {
wcb = new ExistsWatchRegistration(watcher, clientPath);
}
final String serverPath = prependChroot(clientPath);
RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.exists);
ExistsRequest request = new ExistsRequest();
request.setPath(serverPath);
request.setWatch(watcher != null);
SetDataResponse response = new SetDataResponse();
cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
clientPath, serverPath, ctx, wcb);
}
和watcher相關的地方有兩處，一個是其被轉換成WatchRegistration中的一個屬性，最後變成一個Packet（client與server的最小通訊單元）。app
另一處則是request.setWatch(watcher != null)，是個布爾變量。async
ClientCnxn.queuePacket
public Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
Record response, AsyncCallback cb, String clientPath,
String serverPath, Object ctx, WatchRegistration watchRegistration) {
return queuePacket(h, r, request, response, cb, clientPath, serverPath,
ctx, watchRegistration, null);
}
public Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
Record response, AsyncCallback cb, String clientPath,
String serverPath, Object ctx, WatchRegistration watchRegistration,
WatchDeregistration watchDeregistration) {
Packet packet = null;
// Note that we do not generate the Xid for the packet yet. It is
// generated later at send-time, by an implementation of ClientCnxnSocket::doIO(),
// where the packet is actually sent.
packet = new Packet(h, r, request, response, watchRegistration);
packet.cb = cb;
packet.ctx = ctx;
packet.clientPath = clientPath;
packet.serverPath = serverPath;
packet.watchDeregistration = watchDeregistration;
// The synchronized block here is for two purpose:
// 1. synchronize with the final cleanup() in SendThread.run() to avoid race
// 2. synchronized against each packet. So if a closeSession packet is added,
// later packet will be notified.
synchronized (state) {
if (!state.isAlive() || closing) {
conLossPacket(packet);
} else {
// If the client is asking to close the session then
// mark as closing
if (h.getType() == OpCode.closeSession) {
closing = true;
}
outgoingQueue.add(packet);
}
}
sendThread.getClientCnxnSocket().packetAdded();
return packet;
}
這段邏輯簡單來講就是拼裝Packet，並將其加入發送隊列。該隊列由ClientCnxn中的一個SendThread消費（見SendThread.run）。該方法有較多的條件分支，且不夠clean code，故在此再也不貼代碼，避免擾亂視聽。
須要注意的是，WatchRegistration在Packet發送前並不會被序列化發送過去，避免發送沒必要要的信息，畢竟已經在request中標記爲watch了。那麼這個WatchRegistration有什麼用呢？
在Zk的client中，會維護髮送隊列和等待回覆的隊列，裏面都是一個個Packet。
/**
* These are the packets that have been sent and are waiting for a response.
*/
private final LinkedList<Packet> pendingQueue = new LinkedList<Packet>();
/**
* These are the packets that need to be sent.
*/
private final LinkedBlockingDeque<Packet> outgoingQueue = new LinkedBlockingDeque<Packet>();
接下來，咱們查看SendThread.readReponse：
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() == -2) {
// -2 is the xid for pings
if (LOG.isDebugEnabled()) {
LOG.debug("Got ping response for sessionid: 0x"
+ Long.toHexString(sessionId)
+ " after "
+ ((System.nanoTime() - lastPingSentNs) / 1000000)
+ "ms");
}
return;
}
if (replyHdr.getXid() == -4) {
// -4 is the xid for AuthPacket
if(replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) {
state = States.AUTH_FAILED;
eventThread.queueEvent( new WatchedEvent(Watcher.Event.EventType.None,
Watcher.Event.KeeperState.AuthFailed, null) );
eventThread.queueEventOfDeath();
}
if (LOG.isDebugEnabled()) {
LOG.debug("Got auth sessionid:0x"
+ Long.toHexString(sessionId));
}
return;
}
if (replyHdr.getXid() == -1) {
// -1 means notification
if (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 path
if (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;
}
// If SASL authentication is currently in progress, construct and
// send a response packet immediately, rather than queuing a
// response as with other packets.
if (tunnelAuthInProgress()) {
GetSASLRequest request = new GetSASLRequest();
request.deserialize(bbia,"token");
zooKeeperSaslClient.respondToServer(request.getToken(),
ClientCnxn.this);
return;
}
Packet packet;
synchronized (pendingQueue) {
if (pendingQueue.size() == 0) {
throw new IOException("Nothing in the queue, but got "
+ replyHdr.getXid());
}
packet = pendingQueue.remove();
}
/*
* Since requests are processed in order, we better get a response
* to the first request!
*/
try {
if (packet.requestHeader.getXid() != replyHdr.getXid()) {
packet.replyHeader.setErr(
KeeperException.Code.CONNECTIONLOSS.intValue());
throw new IOException("Xid out of order. Got Xid "
+ replyHdr.getXid() + " with err " +
+ replyHdr.getErr() +
" expected Xid "
+ packet.requestHeader.getXid()
+ " for a packet with details: "
+ packet );
}
packet.replyHeader.setXid(replyHdr.getXid());
packet.replyHeader.setErr(replyHdr.getErr());
packet.replyHeader.setZxid(replyHdr.getZxid());
if (replyHdr.getZxid() > 0) {
lastZxid = replyHdr.getZxid();
}
if (packet.response != null && replyHdr.getErr() == 0) {
packet.response.deserialize(bbia, "response");
}
if (LOG.isDebugEnabled()) {
LOG.debug("Reading reply sessionid:0x"
+ Long.toHexString(sessionId) + ", packet:: " + packet);
}
} finally {
finishPacket(packet);
}
}
synchronized (pendingQueue) 中，咱們能夠看到從隊列中拿出了Packet，並最後將其丟入了finishPacket。
protected void finishPacket(Packet p) {
int err = p.replyHeader.getErr();
if (p.watchRegistration != null) {
p.watchRegistration.register(err);
}
// Add all the removed watch events to the event queue, so that the
// clients will be notified with 'Data/Child WatchRemoved' event type.
if (p.watchDeregistration != null) {
Map<EventType, Set<Watcher>> materializedWatchers = null;
try {
materializedWatchers = p.watchDeregistration.unregister(err);
for (Entry<EventType, Set<Watcher>> entry : materializedWatchers
.entrySet()) {
Set<Watcher> watchers = entry.getValue();
if (watchers.size() > 0) {
queueEvent(p.watchDeregistration.getClientPath(), err,
watchers, entry.getKey());
// ignore connectionloss when removing from local
// session
p.replyHeader.setErr(Code.OK.intValue());
}
}
} catch (KeeperException.NoWatcherException nwe) {
p.replyHeader.setErr(nwe.code().intValue());
} catch (KeeperException ke) {
p.replyHeader.setErr(ke.code().intValue());
}
}
if (p.cb == null) {
synchronized (p) {
p.finished = true;
p.notifyAll();
}
} else {
p.finished = true;
eventThread.queuePacket(p);
}
}
能夠肯定的是，WatchDeregistration僅僅維護在client不會發送到server去。register的邏輯很簡單，咱們來看一下：
/**
* Register the watcher with the set of watches on path.
* @param rc the result code of the operation that attempted to
* add the watch on the path.
*/
public void register(int rc) {
if (shouldAddWatch(rc)) {
Map<String, Set<Watcher>> watches = getWatches(rc);
synchronized(watches) {
Set<Watcher> watchers = watches.get(clientPath);
if (watchers == null) {
watchers = new HashSet<Watcher>();
watches.put(clientPath, watchers);
}
watchers.add(watcher);
}
}
}
咱們能夠看到，維護的就是一個path-watchers的字典。接下來來看queueEvent：
void queueEvent(String clientPath, int err,
Set<Watcher> materializedWatchers, EventType eventType) {
KeeperState sessionState = KeeperState.SyncConnected;
if (KeeperException.Code.SESSIONEXPIRED.intValue() == err
|| KeeperException.Code.CONNECTIONLOSS.intValue() == err) {
sessionState = Event.KeeperState.Disconnected;
}
WatchedEvent event = new WatchedEvent(eventType, sessionState,
clientPath);
eventThread.queueEvent(event, materializedWatchers);
}
邏輯很簡單，組裝event，交給eventThread去作通知。
private void queueEvent(WatchedEvent event,
Set<Watcher> materializedWatchers) {
if (event.getType() == EventType.None
&& sessionState == event.getState()) {
return;
}
sessionState = event.getState();
final Set<Watcher> watchers;
if (materializedWatchers == null) {
// materialize the watchers based on the event
watchers = watcher.materialize(event.getState(),
event.getType(), event.getPath());
} else {
watchers = new HashSet<Watcher>();
watchers.addAll(materializedWatchers);
}
WatcherSetEventPair pair = new WatcherSetEventPair(watchers, event);
// queue the pair (watch set & event) for later processing
waitingEvents.add(pair);
}
先看一下ClientWatchManager.materialize。
/* (non-Javadoc)
* @see org.apache.zookeeper.ClientWatchManager#materialize(Event.KeeperState,
* Event.EventType, java.lang.String)
*/
@Override
public Set<Watcher> materialize(Watcher.Event.KeeperState state,
Watcher.Event.EventType type,
String clientPath)
{
Set<Watcher> result = new HashSet<Watcher>();
switch (type) {
case None:
result.add(defaultWatcher);
boolean clear = disableAutoWatchReset && state != Watcher.Event.KeeperState.SyncConnected;
synchronized(dataWatches) {
for(Set<Watcher> ws: dataWatches.values()) {
result.addAll(ws);
}
if (clear) {
dataWatches.clear();
}
}
synchronized(existWatches) {
for(Set<Watcher> ws: existWatches.values()) {
result.addAll(ws);
}
if (clear) {
existWatches.clear();
}
}
synchronized(childWatches) {
for(Set<Watcher> ws: childWatches.values()) {
result.addAll(ws);
}
if (clear) {
childWatches.clear();
}
}
return result;
case NodeDataChanged:
case NodeCreated:
synchronized (dataWatches) {
addTo(dataWatches.remove(clientPath), result);
}
synchronized (existWatches) {
addTo(existWatches.remove(clientPath), result);
}
break;
case NodeChildrenChanged:
synchronized (childWatches) {
addTo(childWatches.remove(clientPath), result);
}
break;
case NodeDeleted:
synchronized (dataWatches) {
addTo(dataWatches.remove(clientPath), result);
}
// XXX This shouldn't be needed, but just in case
synchronized (existWatches) {
Set<Watcher> list = existWatches.remove(clientPath);
if (list != null) {
addTo(list, result);
LOG.warn("We are triggering an exists watch for delete! Shouldn't happen!");
}
}
synchronized (childWatches) {
addTo(childWatches.remove(clientPath), result);
}
break;
default:
String msg = "Unhandled watch event type " + type
+ " with state " + state + " on path " + clientPath;
LOG.error(msg);
throw new RuntimeException(msg);
}
return result;
}
}
注意，當watch被觸發後，即會被移除。client中的ZkWatchManager是維護了watch狀態的：
static class ZKWatchManager implements ClientWatchManager {
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>>();
//......
}
再說回來eventThread.run最後作的事情——又是一個入隊列。那麼咱們來看看這個線程的核心方法：
@Override
@SuppressFBWarnings("JLM_JSR166_UTILCONCURRENT_MONITORENTER")
public void run() {
try {
isRunning = true;
while (true) {
Object event = waitingEvents.take();
if (event == eventOfDeath) {
wasKilled = true;
} else {
processEvent(event);
}
if (wasKilled)
synchronized (waitingEvents) {
if (waitingEvents.isEmpty()) {
isRunning = false;
break;
}
}
}
} catch (InterruptedException e) {
LOG.error("Event thread exiting due to interruption", e);
}
LOG.info("EventThread shut down for session: 0x{}",
Long.toHexString(getSessionId()));
}
接着看processEvent：
private void processEvent(Object event) {
try {
if (event instanceof WatcherSetEventPair) {
// each watcher will process the event
WatcherSetEventPair pair = (WatcherSetEventPair) event;
for (Watcher watcher : pair.watchers) {
try {
watcher.process(pair.event);
} catch (Throwable t) {
LOG.error("Error while calling watcher ", t);
}
}
} else if (event instanceof LocalCallback) {
//在本文中這些邏輯不重要，skip
}
當process被調用後，咱們本身編寫的邏輯就會被觸發啦。
服務端實現
在上文，咱們瞭解了client的watch相關實現，接下來，咱們就來捋一捋服務端的watch實現。
那麼咱們先來看ZkServer handle request的地方，FinalRequestProcessor的processRequest：
case OpCode.exists: {
lastOp = "EXIS";
// TODO we need to figure out the security requirement for this!
ExistsRequest existsRequest = new ExistsRequest();
ByteBufferInputStream.byteBuffer2Record(request.request,
existsRequest);
String path = existsRequest.getPath();
if (path.indexOf('\0') != -1) {
throw new KeeperException.BadArgumentsException();
}
Stat stat = zks.getZKDatabase().statNode(path, existsRequest
.getWatch() ? cnxn : null);
rsp = new ExistsResponse(stat);
break;
}
能夠看到，若是request是要求watch的，那麼會將ServerCnxn傳遞下去——其本質表明了客戶端和服務器之間的鏈接。這樣當數據事件發生時，能夠經過鏈接觸發client的watch。
跳轉DataTreed.statNode：
public Stat statNode(String path, Watcher watcher)
throws KeeperException.NoNodeException {
Stat stat = new Stat();
DataNode n = nodes.get(path);
if (watcher != null) {
dataWatches.addWatch(path, watcher);
}
if (n == null) {
throw new KeeperException.NoNodeException();
}
synchronized (n) {
n.copyStat(stat);
return stat;
}
}
當watcher != null時，則會添加一個watcher當服務端的dataWatches中。接下來，咱們來看一下服務端的watch核心類——WatchManager:
/**
* 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);
private final HashMap<String, HashSet<Watcher>> watchTable =
new HashMap<String, HashSet<Watcher>>();
private final HashMap<Watcher, HashSet<String>> watch2Paths =
new HashMap<Watcher, HashSet<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) {
HashSet<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);
HashSet<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) {
HashSet<String> paths = watch2Paths.remove(watcher);
if (paths == null) {
return;
}
for (String p : paths) {
HashSet<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);
}
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);
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);
}
}
}
for (Watcher w : watchers) {
if (supress != null && supress.contains(w)) {
continue;
}
w.process(e);
}
return watchers;
}
/**
* Brief description of this object.
*/
@Override
public synchronized String toString() {
StringBuilder sb = new StringBuilder();
sb.append(watch2Paths.size()).append(" connections watching ")
.append(watchTable.size()).append(" paths\n");
int total = 0;
for (HashSet<String> paths : watch2Paths.values()) {
total += paths.size();
}
sb.append("Total watches:").append(total);
return sb.toString();
}
/**
* String representation of watches. Warning, may be large!
* @param byPath iff true output watches by paths, otw output
* watches by connection
* @return string representation of watches
*/
synchronized void dumpWatches(PrintWriter pwriter, boolean byPath) {
if (byPath) {
for (Entry<String, HashSet<Watcher>> e : watchTable.entrySet()) {
pwriter.println(e.getKey());
for (Watcher w : e.getValue()) {
pwriter.print("\t0x");
pwriter.print(Long.toHexString(((ServerCnxn)w).getSessionId()));
pwriter.print("\n");
}
}
} else {
for (Entry<Watcher, HashSet<String>> e : watch2Paths.entrySet()) {
pwriter.print("0x");
pwriter.println(Long.toHexString(((ServerCnxn)e.getKey()).getSessionId()));
for (String path : e.getValue()) {
pwriter.print("\t");
pwriter.println(path);
}
}
}
}
/**
* Checks the specified watcher exists for the given path
*
* @param path
* znode path
* @param watcher
* watcher object reference
* @return true if the watcher exists, false otherwise
*/
synchronized boolean containsWatcher(String path, Watcher watcher) {
HashSet<String> paths = watch2Paths.get(watcher);
if (paths == null || !paths.contains(path)) {
return false;
}
return true;
}
/**
* Removes the specified watcher for the given path
*
* @param path
* znode path
* @param watcher
* watcher object reference
* @return true if the watcher successfully removed, false otherwise
*/
synchronized boolean removeWatcher(String path, Watcher watcher) {
HashSet<String> paths = watch2Paths.get(watcher);
if (paths == null || !paths.remove(path)) {
return false;
}
HashSet<Watcher> list = watchTable.get(path);
if (list == null || !list.remove(watcher)) {
return false;
}
if (list.size() == 0) {
watchTable.remove(path);
}
return true;
}
/**
* Returns a watch report.
*
* @return watch report
* @see WatchesReport
*/
synchronized WatchesReport getWatches() {
Map<Long, Set<String>> id2paths = new HashMap<Long, Set<String>>();
for (Entry<Watcher, HashSet<String>> e: watch2Paths.entrySet()) {
Long id = ((ServerCnxn) e.getKey()).getSessionId();
HashSet<String> paths = new HashSet<String>(e.getValue());
id2paths.put(id, paths);
}
return new WatchesReport(id2paths);
}
/**
* Returns a watch report by path.
*
* @return watch report
* @see WatchesPathReport
*/
synchronized WatchesPathReport getWatchesByPath() {
Map<String, Set<Long>> path2ids = new HashMap<String, Set<Long>>();
for (Entry<String, HashSet<Watcher>> e : watchTable.entrySet()) {
Set<Long> ids = new HashSet<Long>(e.getValue().size());
path2ids.put(e.getKey(), ids);
for (Watcher watcher : e.getValue()) {
ids.add(((ServerCnxn) watcher).getSessionId());
}
}
return new WatchesPathReport(path2ids);
}
/**
* Returns a watch summary.
*
* @return watch summary
* @see WatchesSummary
*/
synchronized WatchesSummary getWatchesSummary() {
int totalWatches = 0;
for (HashSet<String> paths : watch2Paths.values()) {
totalWatches += paths.size();
}
return new WatchesSummary (watch2Paths.size(), watchTable.size(),
totalWatches);
}
}
整個類很是好理解，先看兩個核心成員變量：
- watchTable:path-watchs
- watch2Paths:watcher-paths
addWatch就是往兩個map中添加數據，而觸發即是根據path遍歷出那些watcher，並調用它們的process——這時ServerCnxn就會發送一個Packet到client。
那麼何時觸發呢？也很簡單。就在DataTree的代碼裏，對相應數據進行操做時，就會觸發watcher。咱們以DataTree.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));
}
dataWatches.triggerWatch(path, EventType.NodeDataChanged);
return s;
}
至此，咱們就理清watch在Zk裏究竟是怎麼一回事了。這麼一通分析下來，咱們能夠得出watcher的幾個特性：
- 一次性：不管是client仍是server，一旦watcher被觸發，zk都會將其移除。這意味着開發者須要反覆註冊，可是好處也很明顯——下降了服務器壓力，避免頻繁更新的節點一直觸發watcher。
- 客戶端串行執行：客戶端回調是一個串行同步的過程，這使得回調是有序的。一樣，開發者要注意不要由於一個watcher的邏輯影響整個client回調。
-
輕量：client的request是否要watch其實僅僅經過一個boolean來決定，一樣的，server的response的watch回調——WatchedEvent也僅僅只有三個屬性：
- 通知狀態
- 事件類型
- 節點路徑
這種輕量化的設計使得網絡開銷和服務端內存開銷上都是很廉價的。
小結
在本文中，咱們一塊兒瞭解了watch的實現機理。簡單總結以下：
- client在發送請求時候，會將watch的具體狀態保存在client
- 標記watch的request到達服務端後，服務端會將這個watcher以一個字典的形式保存起來
- 當watch的數據發生相應變化時，找出註冊的watch，裏面包含着client的鏈接
- 發送一個通知到client
- client的回調被觸發
相關文章
- 1. 深刻淺出 ZooKeeper
- 2. 深刻淺出Netty源碼剖析
- 3. 轉載：深刻淺出Zookeeper
- 4. 深入淺出剖析SVN
- 5. 深刻剖析HashMap實現原理
- 6. 深刻分析ZooKeeper的實現原理
- 7. 深刻淺出分析LinkedHashMap
- 8. 深刻淺出分析 HashMap
- 9. 深刻淺出Zookeeper（三）：會話管理
- 10. 深刻淺出Zookeeper（七）：Leader選舉
- 更多相關文章...
- • TCP滑動窗口機制深度剖析 - TCP/IP教程
- • 現實生活中的 XML - XML 教程
- • 互聯網組織的未來：剖析GitHub員工的任性之源
- • ☆基於Java Instrument的Agent實現
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章