ZooKeeper source code analysis of the complete network communication process



In the previous chapter, the server handles the request and writes it to the buffer of the client through the sendBuffer() method of NIOServerCnxn. In the previous step, the ConnectResponse response object has been initialized, so RSP will be called after writing Serialize (Bos, "connect") returns a ConnectResponse response object to the client. How to operate when the client receives this response object? See below.

Processing flow after the client receives the response from the server

In the first chapter, the sendThread thread of the client is in the state of waiting for the response of the server when it sends a request. At this time, the server sends back a response, and the sendThread of the client receives the response. At this time, you need to read the contents of the buffer. The specific process is as follows:

(1) because sendThread is in wait state, when CPU calls the thread, the thread executes the run() method, in which clientCnxnSocket. is called. doTransport (to, pendingqueue, clientcnxn. This) is the same as the previous request made by the client, except that the doIO method in doTransport calls different operation statements. The previous request calls write operation, and now the processing response is read operation.

    void doIO(Queue<Packet> pendingQueue, ClientCnxn cnxn) throws InterruptedException, IOException {
        SocketChannel sock = (SocketChannel) sockKey.channel();
        if (sock == null) {
            throw new IOException("Socket is null!");
        // If readable
        if (sockKey.isReadable()) {
            // Read the length of the response packet of the input buffer
            int rc = sock.read(incomingBuffer);
            if (rc < 0) {
                throw new EndOfStreamException("Unable to read additional data from server sessionid 0x"
                                               + Long.toHexString(sessionId)
                                               + ", likely server has closed socket");
            // If the response packet of the input buffer is full, the same packet needs to be read twice, one for length and one for content
            if (!incomingBuffer.hasRemaining()) {
                // If the length is read, allocate space for the packet length to the incomingBuffer
                if (incomingBuffer == lenBuffer) {
                } else if (!initialized) {// If it is not initialized, that is, the session is not established, the server must return ConnectResponse
                    // Read the ConnectRequest and deserialize the contents of incomingBuffer into a ConnectResponse object
                    // Continue reading subsequent responses
                    // If there are write requests, ensure that the write event is ok
                    if (findSendablePacket(outgoingQueue, sendThread.tunnelAuthInProgress()) != null) {
                        // Since SASL authentication has completed (if client is configured to do so),
                        // outgoing packets waiting in the outgoingQueue can now be sent.
                    // Ready to read the next response
                    incomingBuffer = lenBuffer;
                    // Session setup completed
                    initialized = true;
                } else {
                    incomingBuffer = lenBuffer;

(2) Obviously, if the given scenario session is uninitialized at this time, call readConnectResult(), so you need to analyze how this method deserializes the contents of incomingBuffer into ConnectResponse object.

    void readConnectResult() throws IOException {
        if (LOG.isTraceEnabled()) {
            StringBuilder buf = new StringBuilder("0x[");
            for (byte b : incomingBuffer.array()) {
            if (LOG.isTraceEnabled()) {
                LOG.trace("readConnectResult {} {}", incomingBuffer.remaining(), buf.toString());
        // Deserialize incomingBuffer into ConnectResponse
        ByteBufferInputStream bbis = new ByteBufferInputStream(incomingBuffer);
        BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
        ConnectResponse conRsp = new ConnectResponse();
        conRsp.deserialize(bbia, "connect");

        // Read only flag bit
        boolean isRO = false;
        try {
            isRO = bbia.readBool("readOnly");
        } catch (IOException e) {
            // this is ok -- just a packet from an old server which
            // doesn't contain readOnly field
            LOG.warn("Connected to an old server; r-o mode will be unavailable");
        // Session ID returned by the server
        this.sessionId = conRsp.getSessionId();
        // Subsequent processing, initialize some parameters of the client, and finally trigger the WatchedEvent thread
        sendThread.onConnected(conRsp.getTimeOut(), this.sessionId, conRsp.getPasswd(), isRO);

(3) After deserialization, you need to care about subsequent processing. At this time, you need to enter the onConnected method for analysis. onConnected is a specific method under ClientCnxn to initialize some parameters of the client in order to finally trigger the WatchedEvent thread.

        void onConnected(
            int _negotiatedSessionTimeout,
            long _sessionId,
            byte[] _sessionPasswd,
            boolean isRO) throws IOException {
            if (!readOnly && isRO) {
                LOG.error("Read/write client got connected to read-only server");
            // Initialize the session related parameters of the client
            readTimeout = negotiatedSessionTimeout * 2 / 3;
            connectTimeout = negotiatedSessionTimeout / hostProvider.size();
            sessionId = _sessionId;
            sessionPasswd = _sessionPasswd;
            changeZkState((isRO) ? States.CONNECTEDREADONLY : States.CONNECTED);
            seenRwServerBefore |= !isRO;
                "Session establishment complete on server {}, session id = 0x{}, negotiated timeout = {}{}",
                (isRO ? " (READ-ONLY mode)" : ""));
            //A SyncConnected event is triggered, and the EventThread asynchronously notifies the registered watcher to handle it
            KeeperState eventState = (isRO) ? KeeperState.ConnectedReadOnly : KeeperState.SyncConnected;
            eventThread.queueEvent(new WatchedEvent(Watcher.Event.EventType.None, eventState, null));

(4) When the EventThread asynchronously notifies the registered watcher for processing, it will enter the queueEvent for processing.

        private void queueEvent(WatchedEvent event, Set<Watcher> materializedWatchers) {
            if (event.getType() == EventType.None && sessionState == event.getState()) {
            // EventThread sync session state
            sessionState = event.getState();
            final Set<Watcher> watchers;
            if (materializedWatchers == null) {
                // materialize the watchers based on the event
                // Find out those watchers that need to be notified, and the main thread directly calls the corresponding watcher interface
                watchers = watcher.materialize(event.getState(), event.getType(), event.getPath());
            } else {
                watchers = new HashSet<Watcher>();
            WatcherSetEventPair pair = new WatcherSetEventPair(watchers, event);
            // queue the pair (watch set & event) for later processing
            // Commit asynchronous queue processing

(5) At this time, the EventThread thread will execute run(), run to fetch the event from the waitingEvents queue, and call processEvent to process the event.

        public void run() {
            try {
                // Running
                isRunning = true;
                while (true) {
                    // Remove the event handler from the waiting event queue
                    Object event = waitingEvents.take();
                    // If it's a death
                    if (event == eventOfDeath) {
                        // eliminate
                        wasKilled = true;
                    } else {// Otherwise, handle the event
                    // If it's a death
                    if (wasKilled) {
                        // Synchronous processing
                        synchronized (waitingEvents) {
                            if (waitingEvents.isEmpty()) {
                                isRunning = false;
            } catch (InterruptedException e) {
                LOG.error("Event thread exiting due to interruption", e);

(6) Obviously, we are more concerned about how EventThread calls processEvent to handle events. Since processEvent has hundreds of lines of code, here we choose the important ones to explain. In addition, since the watcher is not the main part of this chapter, we will not elaborate here. If you need to know, you can go to this blog to check: https://blog.csdn.net/par_ser?spm=1001.2014.3001.5343

private void processEvent(Object event) {
            try {
                if (event instanceof WatcherSetEventPair) {
                    // Each wathcer handles events
                    WatcherSetEventPair pair = (WatcherSetEventPair) event;
                    for (Watcher watcher : pair.watchers) {
                        try {
                        } catch (Throwable t) {
                            LOG.error("Error while calling watcher ", t);


Here, the network communication process between the client and the server has been analyzed. The session between the client and the server has been established successfully, and then subsequent business processing can be carried out. This paper introduces the network communication process between ZooKeeper client and server, and explains in detail how the client and server interact with data. Although the given scenario is to establish a connection, other network communication business processes are similar to this process.

Keywords: Zookeeper

Added by dumbass on Tue, 21 Dec 2021 14:57:21 +0200