Zookeeper-watcher機制源碼分析（一）
Watcher的基本流程
ZooKeeper 的 Watcher 機制，總的來說可以分為三個過程：客戶端註冊 Watcher、服務器處理 Watcher 和客戶端回調 Watcher
客戶端註冊watcher有3種方式，getData、exists、getChildren；以如下代碼為例來分析整個觸發機制的原理
|
ZooKeeper zookeeper=new ZooKeeper(“192.168.11.152:2181”,4000,new Watcher(){ public void processor(WatchedEvent event){ System.out.println(“event.type”); } })；
zookeeper.create(“/mic”,”0”.getByte(),ZooDefs.Ids. OPEN_ACL_UNSAFE,CreateModel. PERSISTENT); //創建節點
zookeeper.exists(“/mic”,true); //註冊監聽
zookeeper.setData(“/mic”, “1”.getByte(),-1) ; //修改節點的值觸發監聽
|
ZooKeeper API的初始化過程
|
ZooKeeper zookeeper=new ZooKeeper(“192.168.11.152:2181”,4000,new Watcher(){ public void processor(WatchedEvent event){ System.out.println(“event.type”); } })； |
在創建一個 ZooKeeper 客戶端對象實例時，我們通過new Watcher()向構造方法中傳入一個默認的 Watcher, 這個 Watcher 將作為整個 ZooKeeper會話期間的默認 Watcher，會一直被保存在客戶端 ZKWatchManager 的 defaultWatcher 中;代碼如下
|
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly, HostProvider aHostProvider, ZKClientConfig clientConfig) throws IOException { LOG.info("Initiating client connection, connectString=" + connectString + " sessionTimeout=" + sessionTimeout + " watcher=" + watcher);
if (clientConfig == null) { clientConfig = new ZKClientConfig(); } this.clientConfig = clientConfig; watchManager = defaultWatchManager(); watchManager.defaultWatcher = watcher; --在這裏將watcher設置到ZKWatchManager ConnectStringParser connectStringParser = new ConnectStringParser( connectString); hostProvider = aHostProvider; --初始化了ClientCnxn，並且調用cnxn.start()方法 cnxn = new ClientCnxn(connectStringParser.getChrootPath(), hostProvider, sessionTimeout, this, watchManager, getClientCnxnSocket(), canBeReadOnly); cnxn.start(); } |
ClientCnxn:是Zookeeper客戶端和Zookeeper服務器端進行通信和事件通知處理的主要類，它內部包含兩個類，
1. SendThread ：負責客戶端和服務器端的數據通信, 也包括事件信息的傳輸
2. EventThread : 主要在客戶端回調註冊的Watchers進行通知處理
ClientCnxn初始化
|
public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper, ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket, long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) { this.zooKeeper = zooKeeper; this.watcher = watcher; this.sessionId = sessionId; this.sessionPasswd = sessionPasswd; this.sessionTimeout = sessionTimeout; this.hostProvider = hostProvider; this.chrootPath = chrootPath;
connectTimeout = sessionTimeout / hostProvider.size(); readTimeout = sessionTimeout * 2 / 3; readOnly = canBeReadOnly;
sendThread = new SendThread(clientCnxnSocket); --初始化sendThread eventThread = new EventThread(); --初始化eventThread this.clientConfig=zooKeeper.getClientConfig(); }
public void start() { --啟動兩個線程 sendThread.start(); eventThread.start(); } |
客戶端通過exists註冊監聽
|
zookeeper.exists(“/mic”,true); //註冊監聽 |
通過exists方法來註冊監聽，代碼如下
|
public Stat exists(final String path, Watcher watcher) throws KeeperException, InterruptedException { 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); //構建ExistWatchRegistration }
final String serverPath = prependChroot(clientPath);
RequestHeader h = new RequestHeader(); h.setType(ZooDefs.OpCode.exists); //設置操作類型為exists ExistsRequest request = new ExistsRequest(); // 構造ExistsRequest request.setPath(serverPath); request.setWatch(watcher != null); //是否註冊監聽 SetDataResponse response = new SetDataResponse(); //設置服務端響應的接收類 //將封裝的RequestHeader、ExistsRequest、SetDataResponse、WatchRegistration添加到發送隊列 ReplyHeader r = cnxn.submitRequest(h, request, response, wcb); if (r.getErr() != 0) { if (r.getErr() == KeeperException.Code.NONODE.intValue()) { return null; } throw KeeperException.create(KeeperException.Code.get(r.getErr()), clientPath); } //返回exists得到的結果（Stat信息） return response.getStat().getCzxid() == -1 ? null : response.getStat(); } |
cnxn.submitRequest
|
public ReplyHeader submitRequest(RequestHeader h, Record request, Record response, WatchRegistration watchRegistration, WatchDeregistration watchDeregistration) throws InterruptedException { ReplyHeader r = new ReplyHeader(); //將消息添加到隊列,並構造一個Packet傳輸對象 Packet packet = queuePacket(h, r, request, response, null, null, null, null, watchRegistration, watchDeregistration); synchronized (packet) { while (!packet.finished) { //在數據包沒有處理完成之前，一直阻塞 packet.wait(); } } return r; } |
|
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 packet = null; packet = new Packet(h, r, request, response, watchRegistration); packet.cb = cb; packet.ctx = ctx; packet.clientPath = clientPath; packet.serverPath = serverPath; packet.watchDeregistration = watchDeregistration;
synchronized (state) { if (!state.isAlive() || closing) { conLossPacket(packet); } else { if (h.getType() == OpCode.closeSession) { closing = true; } outgoingQueue.add(packet); //添加到outgoingQueue } } sendThread.getClientCnxnSocket().packetAdded();//此處是多路復用機制，喚醒Selector，告訴他有數據包添加過來了 return packet; } |
在 ZooKeeper 中，Packet 是一個最小的通信協議單元，即數據包。Pakcet 用於進行客戶端與服務端之間的網絡傳輸，任何需要傳輸的對象都需要包裝成一個 Packet 對象。在 ClientCnxn 中 WatchRegistration 也會被封裝到 Pakcet 中，然後由 SendThread 線程調用queuePacket方法把 Packet 放入發送隊列中等待客戶端發送，這又是一個異步過程，分布式系統采用異步通信是一個非常常見的手段
SendThread的發送過程
在初始化連接的時候，zookeeper初始化了兩個線程並且啟動了。接下來我們來分析SendThread的發送過程，因為是一個線程，所以啟動的時候會調用SendThread.run方法
|
public void run() { clientCnxnSocket.introduce(this, sessionId, outgoingQueue); clientCnxnSocket.updateNow(); clientCnxnSocket.updateLastSendAndHeard(); int to; long lastPingRwServer = Time.currentElapsedTime(); final int MAX_SEND_PING_INTERVAL = 10000; //10 seconds while (state.isAlive()) { try { if (!clientCnxnSocket.isConnected()) {// 如果沒有連接：發起連接 // don‘t re-establish connection if we are closing if (closing) { break; } startConnect(); //發起連接 clientCnxnSocket.updateLastSendAndHeard(); }
if (state.isConnected()) { //如果是連接狀態，則處理sasl的認證授權 // determine whether we need to send an AuthFailed event. if (zooKeeperSaslClient != null) { boolean sendAuthEvent = false; if (zooKeeperSaslClient.getSaslState() == ZooKeeperSaslClient.SaslState.INITIAL) { try { zooKeeperSaslClient.initialize(ClientCnxn.this); } catch (SaslException e) { LOG.error("SASL authentication with Zookeeper Quorum member failed: " + e); state = States.AUTH_FAILED; sendAuthEvent = true; } } KeeperState authState = zooKeeperSaslClient.getKeeperState(); if (authState != null) { if (authState == KeeperState.AuthFailed) { // An authentication error occurred during authentication with the Zookeeper Server. state = States.AUTH_FAILED; sendAuthEvent = true; } else { if (authState == KeeperState.SaslAuthenticated) { sendAuthEvent = true; } } }
if (sendAuthEvent == true) { eventThread.queueEvent(new WatchedEvent( Watcher.Event.EventType.None, authState,null)); } } to = readTimeout - clientCnxnSocket.getIdleRecv(); } else { to = connectTimeout - clientCnxnSocket.getIdleRecv(); } //to,表示客戶端距離timeout還剩多少時間，準備發起ping連接 if (to <= 0) {//表示已經超時了。 String warnInfo; warnInfo = "Client session timed out, have not heard from server in " + clientCnxnSocket.getIdleRecv() + "ms" + " for sessionid 0x" + Long.toHexString(sessionId); LOG.warn(warnInfo); throw new SessionTimeoutException(warnInfo); } if (state.isConnected()) { //計算下一次ping請求的時間 int timeToNextPing = readTimeout / 2 - clientCnxnSocket.getIdleSend() - ((clientCnxnSocket.getIdleSend() > 1000) ? 1000 : 0); //send a ping request either time is due or no packet sent out within MAX_SEND_PING_INTERVAL if (timeToNextPing <= 0 || clientCnxnSocket.getIdleSend() > MAX_SEND_PING_INTERVAL) { sendPing(); //發送ping請求 clientCnxnSocket.updateLastSend(); } else { if (timeToNextPing < to) { to = timeToNextPing; } } } // If we are in read-only mode, seek for read/write server if (state == States.CONNECTEDREADONLY) { long now = Time.currentElapsedTime(); int idlePingRwServer = (int) (now - lastPingRwServer); if (idlePingRwServer >= pingRwTimeout) { lastPingRwServer = now; idlePingRwServer = 0; pingRwTimeout = Math.min(2*pingRwTimeout, maxPingRwTimeout); pingRwServer(); } to = Math.min(to, pingRwTimeout - idlePingRwServer); } 調用clientCnxnSocket，發起傳輸 其中 pendingQueue是一個用來存放已經發送、等待回應的Packet隊列， clientCnxnSocket默認使用ClientCnxnSocketNIO（ps：還記得在哪裏初始化嗎？在實例化zookeeper的時候） clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this); } catch (Throwable e) { if (closing) { if (LOG.isDebugEnabled()) { // closing so this is expected LOG.debug("An exception was thrown while closing send thread for session 0x" + Long.toHexString(getSessionId()) + " : " + e.getMessage()); } break; } else { // this is ugly, you have a better way speak up if (e instanceof SessionExpiredException) { LOG.info(e.getMessage() + ", closing socket connection"); } else if (e instanceof SessionTimeoutException) { LOG.info(e.getMessage() + RETRY_CONN_MSG); } else if (e instanceof EndOfStreamException) { LOG.info(e.getMessage() + RETRY_CONN_MSG); } else if (e instanceof RWServerFoundException) { LOG.info(e.getMessage()); } else { LOG.warn( "Session 0x" + Long.toHexString(getSessionId()) + " for server " + clientCnxnSocket.getRemoteSocketAddress() + ", unexpected error" + RETRY_CONN_MSG, e); } // At this point, there might still be new packets appended to outgoingQueue. // they will be handled in next connection or cleared up if closed. cleanup(); if (state.isAlive()) { eventThread.queueEvent(new WatchedEvent( Event.EventType.None, Event.KeeperState.Disconnected, null)); } clientCnxnSocket.updateNow(); clientCnxnSocket.updateLastSendAndHeard(); } } } synchronized (state) { // When it comes to this point, it guarantees that later queued // packet to outgoingQueue will be notified of death. cleanup(); } clientCnxnSocket.close(); if (state.isAlive()) { eventThread.queueEvent(new WatchedEvent(Event.EventType.None, Event.KeeperState.Disconnected, null)); } ZooTrace.logTraceMessage(LOG, ZooTrace.getTextTraceLevel(), "SendThread exited loop for session: 0x" + Long.toHexString(getSessionId())); } |
client 和 server的網絡交互
|
@Override void doTransport(int waitTimeOut, List<Packet> pendingQueue, ClientCnxn cnxn) throws IOException, InterruptedException { try { if (!firstConnect.await(waitTimeOut, TimeUnit.MILLISECONDS)) { return; } Packet head = null; if (needSasl.get()) { if (!waitSasl.tryAcquire(waitTimeOut, TimeUnit.MILLISECONDS)) { return; } } else { //判斷outgoingQueue是否存在待發送的數據包，不存在則直接返回 if ((head = outgoingQueue.poll(waitTimeOut, TimeUnit.MILLISECONDS)) == null) { return; } } // check if being waken up on closing. if (!sendThread.getZkState().isAlive()) { // adding back the patck to notify of failure in conLossPacket(). addBack(head); return; } // channel disconnection happened if (disconnected.get()) { //異常流程，channel關閉了，講當前的packet添加到addBack中 addBack(head); throw new EndOfStreamException("channel for sessionid 0x" + Long.toHexString(sessionId) + " is lost"); } if (head != null) { //如果當前存在需要發送的數據包，則調用doWrite方法，pendingQueue表示處於已經發送過等待響應的packet隊列 doWrite(pendingQueue, head, cnxn); } } finally { updateNow(); } } |
DoWrite方法
|
private void doWrite(List<Packet> pendingQueue, Packet p, ClientCnxn cnxn) { updateNow(); while (true) { if (p != WakeupPacket.getInstance()) { if ((p.requestHeader != null) && //判斷請求頭以及判斷當前請求類型不是ping或者auth操作 (p.requestHeader.getType() != ZooDefs.OpCode.ping) && (p.requestHeader.getType() != ZooDefs.OpCode.auth)) { p.requestHeader.setXid(cnxn.getXid()); //設置xid，這個xid用來區分請求類型 synchronized (pendingQueue) { pendingQueue.add(p); //將當前的packet添加到pendingQueue隊列中 } } sendPkt(p); //將數據包發送出去 } if (outgoingQueue.isEmpty()) { break; } p = outgoingQueue.remove(); } } |
sendPkt
|
private void sendPkt(Packet p) { // Assuming the packet will be sent out successfully. Because if it fails, // the channel will close and clean up queues. p.createBB(); //序列化請求數據 updateLastSend(); //更新最後一次發送updateLastSend sentCount++; //更新發送次數 channel.write(ChannelBuffers.wrappedBuffer(p.bb)); //通過nio channel發送字節緩存到服務端 } |
createBB
|
public void createBB() { try { ByteArrayOutputStream baos = new ByteArrayOutputStream(); BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos); boa.writeInt(-1, "len"); // We‘ll fill this in later if (requestHeader != null) { requestHeader.serialize(boa, "header"); //序列化header頭(requestHeader) } if (request instanceof ConnectRequest) { request.serialize(boa, "connect"); // append "am-I-allowed-to-be-readonly" flag boa.writeBool(readOnly, "readOnly"); } else if (request != null) { request.serialize(boa, "request"); //序列化request(request) } baos.close(); this.bb = ByteBuffer.wrap(baos.toByteArray()); this.bb.putInt(this.bb.capacity() - 4); this.bb.rewind(); } catch (IOException e) { LOG.warn("Ignoring unexpected exception", e); } } |
從createBB方法中，我們看到在底層實際的網絡傳輸序列化中，zookeeper只會講requestHeader和request兩個屬性進行序列化，即只有這兩個會被序列化到底層字節數組中去進行網絡傳輸，不會將watchRegistration相關的信息進行網絡傳輸。
總結
用戶調用exists註冊監聽以後，會做幾個事情
- 講請求數據封裝為packet，添加到outgoingQueue
- SendThread這個線程會執行數據發送操作，主要是將outgoingQueue隊列中的數據發送到服務端
- 通過clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this); 其中ClientCnxnSocket只zookeeper客戶端和服務端的連接通信的封裝，有兩個具體的實現類ClientCnxnSocketNetty和ClientCnxnSocketNIO;具體使用哪一個類來實現發送，是在初始化過程是在實例化Zookeeper的時候設置的，代碼如下
|
cnxn = new ClientCnxn(connectStringParser.getChrootPath(), hostProvider, sessionTimeout, this, watchManager, getClientCnxnSocket(), canBeReadOnly); |
private ClientCnxnSocket getClientCnxnSocket() throws IOException { String clientCnxnSocketName = getClientConfig().getProperty( ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET); if (clientCnxnSocketName == null) { clientCnxnSocketName = ClientCnxnSocketNIO.class.getName(); } try { Constructor<?> clientCxnConstructor = Class.forName(clientCnxnSocketName).getDeclaredConstructor(ZKClientConfig.class); ClientCnxnSocket clientCxnSocket = (ClientCnxnSocket) clientCxnConstructor.newInstance(getClientConfig()); return clientCxnSocket; } catch (Exception e) { IOException ioe = new IOException("Couldn‘t instantiate " + clientCnxnSocketName); ioe.initCause(e); throw ioe; } } |
4.基於第3步，最終會在ClientCnxnSocketNetty方法中執行sendPkt將請求的數據包發送到服務端
對Java技術，架構技術感興趣的同學，歡迎加QQ群619881427，一起學習，相互討論。
群內已經有小夥伴將知識體系整理好（源碼，筆記，PPT，學習視頻），歡迎加群免費領取。
分享給喜歡Java，喜歡編程，有夢想成為架構師的程序員們，希望能夠幫助到你們。
不是Java程序員也沒關系，幫忙轉發給更多朋友！謝謝。
分享一個小技巧點擊閱讀原文也可以輕松獲取到學習資料哦！！
Zookeeper-watcher機制源碼分析（一）