Watcher-- 數(shù)據(jù)變更通知

• 我們知道Zookeeper提供來分布式數(shù)據(jù)的訂閱/發(fā)布功能，一個典型的發(fā)布/訂閱模型系統(tǒng)定義了一種一對多的訂閱關(guān)系，能讓多個訂閱者同時監(jiān)聽某個主題對象，當(dāng)這個被監(jiān)聽對象自身狀態(tài)發(fā)生變化時候，會通知所有訂閱者，Zookeeper中引入了Watcher機(jī)制來實(shí)現(xiàn)這種分布式通知功能，Zookeeper允許客戶端向服務(wù)器節(jié)點(diǎn)注冊一個Watcher監(jiān)聽，當(dāng)服務(wù)器端節(jié)點(diǎn)發(fā)生指定觸發(fā)的事件就會觸發(fā)這個Watcher，之后服務(wù)端會向指定客戶端發(fā)送一個事件通知，這樣來實(shí)現(xiàn)一個分布式通知的功能，如下圖所示的一個流程：
  
• 上圖中流程，Zookeeper的Watcher機(jī)制主要包括客戶端線程，客戶端WatcherManager，和Zookeeper服務(wù)器，流程上簡單的說：
  • 客戶端向Zookeeper服務(wù)器注冊成功Watcher同時，將Watcher對象存儲在客戶端的WatcherManager
  • 當(dāng)Zookeeper服務(wù)器觸發(fā)Watcher事件后，向客戶端發(fā)送通知
  • 客戶端線程從WatcherManager中撈出對應(yīng)的Watcher對象來執(zhí)行回調(diào)邏輯

Watcher接口

• 在Zookeeper中，接口Watcher表示一個標(biāo)準(zhǔn)的事件處理器，訂閱來通知相關(guān)的邏輯，我們可以看他的源碼：
  • EventType：事件類型
  • KeeperState：通知狀態(tài)
  • Process(WatchedEvent event)：會調(diào)方法
• 其中事件類型和通知狀態(tài)是有對應(yīng)關(guān)系，如下表中所示

KeeperStateEventType觸發(fā)條件解釋說明

SyncConnected	None	客戶端與服務(wù)器成功建立連接	客戶端和服務(wù)器處于連接狀態(tài)
SyncConnected	NodeCreated	Watcher 監(jiān)聽的對應(yīng)數(shù)據(jù)節(jié)點(diǎn)成功創(chuàng)建	客戶端和服務(wù)器處于連接狀態(tài)
SyncConnected	NodeDeleted	Watcher監(jiān)聽的數(shù)據(jù)節(jié)點(diǎn)成功刪除	客戶端和服務(wù)器處于連接狀態(tài)
SyncConnected	NodeDataChanged	Watcher監(jiān)聽的數(shù)據(jù)節(jié)點(diǎn)內(nèi)容變更	客戶端和服務(wù)器處于連接狀態(tài)
SyncConnected	NodeChildrenChanged	Watcher監(jiān)聽的對應(yīng)數(shù)據(jù)節(jié)點(diǎn)列表發(fā)生變更	客戶端和服務(wù)器處于連接狀態(tài)
Disconnected	None	客戶端與Zookeeper服務(wù)器斷開連接	客戶端和服務(wù)器斷開了連接
Expired	None	會話超時	客戶端回話失效，通常同時也會收到SessionExpiredException異常
AuthFailed	None	兩種情況：使用錯誤scheme進(jìn)行權(quán)限檢查， SASL權(quán)限檢查失敗	通常同時收到AuthFailedException異常
Unknown			3.1.0后廢棄
NoSYncConnected			3.1.0后廢棄

• 如上列舉了Zookeeper中常見的幾個通知狀態(tài)和事件類型，其中針對NodeDateChange事件說明的節(jié)點(diǎn)的變更并不一定是內(nèi)容變化，可能版本號DataVersion變化也是一樣會觸發(fā)。

• 回調(diào)方法process 是Watcher接口中的一個回調(diào)方法，當(dāng)Zookeeper服務(wù)器端向客戶端發(fā)送一個Watcher事件通知的時候，客戶端會對相應(yīng)的Process方法進(jìn)行回調(diào)，從而實(shí)現(xiàn)對事件的處理，Process方法定義如下

void process(WatchedEvent event);

• 如上參數(shù)WatcherEvent包含了一個事件的基本屬性：

public class WatchedEvent {private final KeeperState keeperState; //通知狀態(tài)private final EventType eventType; // 事件類型private String path; // 節(jié)點(diǎn)路徑/*** Create a WatchedEvent with specified type, state and path*/public WatchedEvent(EventType eventType, KeeperState keeperState, String path) {this.keeperState = keeperState;this.eventType = eventType;this.path = path;} ...... }

• Zookeeper服務(wù)端生成WatchedEvent事件后會調(diào)用getWrapper方法將字節(jié)包裝成一個可序列化的WatcherEvent，其實(shí)這是一個事務(wù)，都是對服務(wù)端事件的一個封裝，不同的是WatchedEvent是我們邏輯事件中的一個對象，主要用來我們程序內(nèi)部的事件容器，而WatcherEvent因?yàn)閷?shí)現(xiàn)了序列化的接口，因此可以用于網(wǎng)絡(luò)傳輸
• 在服務(wù)端得到WatcherEvent后，通過網(wǎng)絡(luò)傳到客戶端，還原成一個WatchedEvent，并傳遞給process，然后process方法根據(jù)入?yún)⒕涂梢越馕鐾暾姆?wù)端事件了。

工作機(jī)制

• Zookeeper的Watcher機(jī)制可以有如下三個過程：
  • 客戶端注冊Watcher
  • 服務(wù)端處理watcher
  • 客戶端回調(diào)Watcher
• 以下類圖說明各組件之間的關(guān)系：
  

客戶端注冊Watcher

• 我們通過如下部分源碼來分析Watcher的客戶端注冊，我們創(chuàng)建一個Zookeeper的客戶端對象實(shí)例時，可以向構(gòu)造方法中傳入一個默認(rèn)的Watcher：

//我們調(diào)用的方法 public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException {this(connectString, sessionTimeout, watcher, false); } //實(shí)際上初始化的方法 public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly, HostProvider aHostProvider, ZKClientConfig clientConfig) throws IOException {LOG.info("Initiating client connection, connectString={} sessionTimeout={} watcher={}", new Object[]{connectString, Integer.valueOf(sessionTimeout), watcher});if(clientConfig == null) {clientConfig = new ZKClientConfig();}this.clientConfig = clientConfig;this.watchManager = this.defaultWatchManager();this.watchManager.defaultWatcher = watcher;ConnectStringParser connectStringParser = new ConnectStringParser(connectString);this.hostProvider = aHostProvider;this.cnxn = this.createConnection(connectStringParser.getChrootPath(), this.hostProvider, sessionTimeout, this, this.watchManager, this.getClientCnxnSocket(), canBeReadOnly);this.cnxn.start();}

• 如上源碼中我們給定的Watcher對象實(shí)際上被保存在客戶端ZKWatcherManager的defaultWatcher中，另外Zookeeper客戶端也可以通過getData，getChildren，exist三個接口來向Zookeeper服務(wù)器注冊Watcher，無論哪一種都一樣，我們用getData方法的源碼來分析：

public byte[] getData(String path, Watcher watcher, Stat stat) throws KeeperException, InterruptedException {PathUtils.validatePath(path); //校驗(yàn)Path格式正確性ZooKeeper.DataWatchRegistration wcb = null;if(watcher != null) {wcb = new ZooKeeper.DataWatchRegistration(watcher, path);//封裝DataWatchRegistration}String serverPath = this.prependChroot(path);RequestHeader h = new RequestHeader();h.setType(4);GetDataRequest request = new GetDataRequest();request.setPath(serverPath);request.setWatch(watcher != null);GetDataResponse response = new GetDataResponse();ReplyHeader r = this.cnxn.submitRequest(h, request, response, wcb);if(r.getErr() != 0) {throw KeeperException.create(Code.get(r.getErr()), path);} else {if(stat != null) {DataTree.copyStat(response.getStat(), stat);}return response.getData();}}

• 如上源碼中參數(shù)Path， Watcher對象，getData接口注冊Watcher后，做了兩件事情
  • 先用這兩個參數(shù)封裝來一個DataWatchRegistration，其實(shí)就是初始化來Zookeeper服務(wù)器中的WatchRegistration里面的 watcher，clientPath，這部分用來暫時存儲注冊信息保存節(jié)點(diǎn)和Watcher的對應(yīng)關(guān)系
  • 接著會向客戶端請求request進(jìn)行標(biāo)記，將其設(shè)置為“使用watcher監(jiān)聽”。
• 接著繼續(xù)往下SubmitRequest方法：

ReplyHeader r = this.cnxn.submitRequest(h, request, response, wcb); public ReplyHeader submitRequest(RequestHeader h,Record request,Record response,WatchRegistration watchRegistration,WatchDeregistration watchDeregistration) throws InterruptedException {ReplyHeader r = new ReplyHeader();Packet packet = queuePacket(h,r,request,response,null, null,null, null, watchRegistration, watchDeregistration);......return r;}

• 這個步驟中又一次將ClientCnxn中的WatchRegistration封裝到Packet中，Zookeeper中，Packet可以被看作是一個最小通信協(xié)議單元，用于進(jìn)行客戶端與服務(wù)器之間的網(wǎng)絡(luò)傳輸，任何需要傳輸?shù)膶ο蠖夹枰b成一個Packet對象，接著他被放入發(fā)送隊(duì)列，如下queuePacket代碼：

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) {......outgoingQueue.add(packet);....}

• 我們繼續(xù)追這個outgoingQueue 隊(duì)列，可以看到隨后Zookeeper客戶端會向服務(wù)器端發(fā)送這個請求，同時等待請求的返回，王朝請求發(fā)送后，會由客戶端的SendThread線程的readResponse方法負(fù)責(zé)接受來自服務(wù)端的響應(yīng)，finishPacket方法會從Packet中取出對于的Watcher并注冊到ZKWatcherManager中去。

protected void finishPacket(Packet p) {int err = p.replyHeader.getErr();if (p.watchRegistration != null) {p.watchRegistration.register(err);}......

• 如Packet中的Watchregistration就是我們剛才第一步getData中保存的節(jié)點(diǎn)對應(yīng)的Watcher的注冊信息。現(xiàn)在他又從這部分中取出來封裝的Watcher，如下具體的register方法：

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);}}}//getWatchesprotected final ZKWatchManager watchManager;protected Map<String, Set<Watcher>> getWatches(int rc) {return watchManager.dataWatches;}

• 如上register方法中客戶端將之前暫時保存的Watcher取出來之后，放入到getWatcher獲取到的一個Map對象中，這個Mp對象就是ZkWatcherManager中的一個dataWatches，我們將剛才存入WatchRegistration中的臨時信息取出用來初始化ZKWatchManager.dataWatches，用于將數(shù)據(jù)節(jié)點(diǎn)的路徑和watcher對象進(jìn)行一一映射，這樣就完成來客戶端Watcher的注冊，整個Watcher流程如下
  

• 如上流程中我們每次調(diào)用getData都會注冊一個Watcher，如果這些Watcher都隨著請求發(fā)送到服務(wù)器的話肯定會內(nèi)存緊張，現(xiàn)實(shí)是這樣的碼，我們可以看之前代碼中負(fù)責(zé)傳輸數(shù)據(jù)的對象Packet中，我們將WatchRegistration封裝進(jìn)去，如下Packet中的序列化方法createBB:

public void createBB() {try {ByteArrayOutputStream baos = new ByteArrayOutputStream();BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);boa.writeInt(-1, "len"); // We'll fill this in laterif (requestHeader != null) {requestHeader.serialize(boa, "header");//封裝requestHeader}if (request instanceof ConnectRequest) {//封裝requestrequest.serialize(boa, "connect");// append "am-I-allowed-to-be-readonly" flagboa.writeBool(readOnly, "readOnly");} else if (request != null) {request.serialize(boa, "request");}baos.close();this.bb = ByteBuffer.wrap(baos.toByteArray());this.bb.putInt(this.bb.capacity() - 4);this.bb.rewind();} catch (IOException e) {LOG.warn("Unexpected exception", e);}}

• 如上源碼中可以看到并沒有整個對象完全序列化進(jìn)去，zookeeper只是將requestHeader和request兩個屬性進(jìn)行序列化，WatchRegistration并沒有被序列化到底層字節(jié)數(shù)組中，所以不會進(jìn)行網(wǎng)絡(luò)傳輸

服務(wù)端處理Watcher

• 上面講解了客戶端注冊Watcher的過程，并且已經(jīng)了解了最終客戶端不會將Watcher對象真正床底到服務(wù)器，那么，服務(wù)端是怎么樣完成客戶端的Watcher注冊，一下我們對這部分文件進(jìn)行解析。

ServerCnxn存儲

• 我們先看下服務(wù)器接收Watcher并將其存儲起來的過程，如下Zookeeper服務(wù)端處理Watcher序列圖：
  
• 我們先從源頭分析客戶端給了服務(wù)器那些信息，如下Zookeeper類中g(shù)etData方法的源碼：

...... RequestHeader h = new RequestHeader(); h.setType(ZooDefs.OpCode.getData); GetDataRequest request = new GetDataRequest(); request.setPath(serverPath); request.setWatch(watcher != null); GetDataResponse response = new GetDataResponse(); ......

• 如上RequestHeader中type類型設(shè)置的 4 ，request中給定了節(jié)點(diǎn)path路徑，以及一個boolean類型的watcher標(biāo)識是否天劍監(jiān)聽。服務(wù)端收到來自客戶端的請求后，在FinalRequestProcessor.processRequest()中會判斷當(dāng)前請求的類型type來做一個策略來決定處理不同類型的請求，如下源碼：

switch (request.type) { ......case OpCode.getData: {......Stat stat = new Stat();byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat,getDataRequest.getWatch() ? cnxn : null);//為tru則會傳遞一個 ServerCnxn cnxn = request.cnxn;對象到實(shí)際的注冊方法中，否則給nullrsp = new GetDataResponse(b, stat);break;} ......}

• 如上，從getData請求的處理邏輯可以看到當(dāng)getDataRequest.getwatch為true的時候，Zookeeper就認(rèn)為當(dāng)前客戶端請求需要進(jìn)行Watcher注冊，于是將當(dāng)前的ServerCnxn對象和數(shù)據(jù)節(jié)點(diǎn)路徑傳入getData方法
• ServerCnxn是一個Zookeeper客戶端和服務(wù)器之間的鏈接接口，代表了一個客戶端和服務(wù)器的鏈接，ServerCnxn接口默認(rèn)實(shí)現(xiàn)是NIOServerCNxn，同時3.4.0版本開始引入了Netty實(shí)現(xiàn)：NettyServerCnxn，都實(shí)現(xiàn)了Watcher接口并且實(shí)現(xiàn)process接口，所有把他看成一個Watcher對象，如下ServerCnxn對象以及兩種process實(shí)現(xiàn)

public abstract class ServerCnxn implements Stats, Watcher { ......public abstract void process(WatchedEvent event);...... }

• 繼續(xù)追getData源碼，getZkDataBase獲取到的ZKDatabase 對象，其中DataTree 對象是現(xiàn)在Zookeeper現(xiàn)有的節(jié)點(diǎn)數(shù)據(jù)的樹形存儲，我們可以通過path來從這獲取到對應(yīng)節(jié)點(diǎn)信息，如下獲取DataNode，初始化節(jié)點(diǎn)狀態(tài)，將DataNode天驕到WatchManager 對象中的WatchTable和watch2Paths中

byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat,getDataRequest.getWatch() ? cnxn : null); //如下getData實(shí)現(xiàn) public byte[] getData(String path, Stat stat, Watcher watcher)throws KeeperException.NoNodeException {DataNode n = nodes.get(path);if (n == null) {throw new KeeperException.NoNodeException();}synchronized (n) {n.copyStat(stat);if (watcher != null) {dataWatches.addWatch(path, watcher);}return n.data;}}

• Watchmanager是Zookeeper服務(wù)端Watcher的管理者，內(nèi)部管理的WatcherTable和Watch2Paths,所以一個節(jié)點(diǎn)存儲了兩次，不過是從如下兩個未存存儲
  • watchTable是從數(shù)據(jù)節(jié)點(diǎn)路徑的粒度來托管Watcher
  • watch2Paths是從Watcher的粒度來空值時間觸發(fā)需要出發(fā)的數(shù)據(jù)節(jié)點(diǎn)。

*/ public 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>>();......}

• WatcherManager數(shù)據(jù)結(jié)構(gòu)如下

WatcherManager

- watchTable: HashMap<String, HashSet>(); + watch2Paths :new HashMap<Watcher, HashSet>();

+ addwatch(String ,Watcher): void + removeWatcher(Watcher): void + triggerWatch(String, EventType):Set +Trigger

上一篇Zookeeper–ZAB與Paxos算法聯(lián)系與區(qū)別
下一篇Zookeeper–Watcher機(jī)制源碼剖析二

總結(jié)

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

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