zookeeper客户端会话状态分析

前言：

Zookeeper客户端创建完成之后，后续就可以使用其进行请求发送，在请求发送之前，会先检查客户端与服务端的连接是否存在（是否有对应Session），如果不存在，则会先创建Session会话，后续的操作都会依据当前Session来发送。

而当连接发生异常时（网络波动等），客户端会尝试重连服务端，其中又会涉及到Session状态的变化。本文就来了解下Session的创建及其状态的变化过程。

1.客户端连接的状态

客户端与服务端三次握手完成后，会发送ConnectRequest请求，服务端处理完成后，返回ConnectResponse，代表当前Session会话创建完成，会话里面有一个sessionId，代表当前会话（很关键）。

那么连接具体有哪些状态呢？具体都写在Zookeeper.States里，具体代码如下

public enum States {
    CONNECTING, ASSOCIATING, CONNECTED, CONNECTEDREADONLY,
    CLOSED, AUTH_FAILED, NOT_CONNECTED;
    // 当前连接是否存活
    public boolean isAlive() {
        return this != CLOSED && this != AUTH_FAILED;
    }
    // 当前连接是否连接成功
    public boolean isConnected() {
        return this == CONNECTED || this == CONNECTEDREADONLY;
    }
}

状态还是比较多的，下面我们来看下这个状态的变化过程。

2.NOT_CONNECTED 未连接

未连接状态是默认状态，具体见ClientCnxn

volatile States state = States.NOT_CONNECTED;

3.CONNECTING 连接中

CONNECTING代表还未连接到Zookeeper服务端，在以下场景中会被设置

3.1 SendThread新建

SendThread(ClientCnxnSocket clientCnxnSocket) {
    super(makeThreadName("-SendThread()"));
    // 当SendThread新建时会将ClientCnxn 的连接状态设置为CONNECTING
    state = States.CONNECTING;
    this.clientCnxnSocket = clientCnxnSocket;
    setDaemon(true);
}

**3.2 ClientCnxn.startConnect() **

public class ClientCnxn {
    private void startConnect(InetSocketAddress addr) throws IOException {
            state = States.CONNECTING;
            ...
            logStartConnect(addr);
            clientCnxnSocket.connect(addr);
        }
}

在客户端发送请求之前，若检测到与服务端之间的连接未创建，则会先发起创建连接的请求；

若发生网络异常时，也会重新创建连接。

4.CONNECTED 、CONNECTEDREADONLY 连接已完成

当成功接收到服务端对于ConnectRequest请求的响应（ConnectResponse）时，则会设置当前连接状态为已完成，具体代码如下

public class ClientCnxnSocketNIO extends ClientCnxnSocket {
    void doIO(List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue, ClientCnxn cnxn)
      throws InterruptedException, IOException {
        SocketChannel sock = (SocketChannel) sockKey.channel();
        if (sockKey.isReadable()) {
            int rc = sock.read(incomingBuffer);
            ...
            if (!incomingBuffer.hasRemaining()) {
                incomingBuffer.flip();
                if (incomingBuffer == lenBuffer) {
                    recvCount++;
                    readLength();
                } else if (!initialized) {
                    // 如果发现还未初始化完成，说明当前响应为连接的响应
                    readConnectResult();
                    ...
                }
            }
        }
    }
    
    void readConnectResult() throws IOException {
        // 处理连接响应
        ByteBufferInputStream bbis = new ByteBufferInputStream(incomingBuffer);
        BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
        ConnectResponse conRsp = new ConnectResponse();
        conRsp.deserialize(bbia, "connect");
        // 客户端的readonly标志
        boolean isRO = false;
        try {
            isRO = bbia.readBool("readOnly");
        } catch (IOException e) {
            LOG.warn("Connected to an old server; r-o mode will be unavailable");
        }
        this.sessionId = conRsp.getSessionId();
        // 继续交由SendThread执行
        sendThread.onConnected(conRsp.getTimeOut(), this.sessionId,
                conRsp.getPasswd(), isRO);
    }
}

SendThread.onConnected()

创建连接完成后，则当前会话状态变成CONNECTED或CONNECTEDREADONLY，这个具体由请求的Packet.readOnly标识

5.AUTH_FAILED 验证失败

当服务端连接需要用户名密码时，那么客户端连接的创建需要将正确的用户名密码传入，如果错误的话，服务端会返回该异常。具体代码如下

class SendThread extends ZooKeeperThread {
        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() == -4) {
                // 代表验证失败         
                if(replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) {
                    state = States.AUTH_FAILED;                    
                    eventThread.queueEvent( new WatchedEvent(Watcher.Event.EventType.None, 
                            Watcher.Event.KeeperState.AuthFailed, null) );                                        
                }
                
                return;
            }
            ...
        }
}

6.CLOSED 关闭连接

当出现会话超时或者客户端主动退出时，那么会话的状态则直接变为CLOSED，具体场景如下

6.1 会话超时时间异常

class SendThread extends ZooKeeperThread {
    void onConnected(int _negotiatedSessionTimeout, long _sessionId,
                byte[] _sessionPasswd, boolean isRO) throws IOException {
            negotiatedSessionTimeout = _negotiatedSessionTimeout;
        // 如果协商出的超时时间<=0，那么直接关闭当前连接
            if (negotiatedSessionTimeout <= 0) {
                state = States.CLOSED;
            }
    }
}

6.2 ClinetCnxn.close() 客户端主动退出

public class ClientCnxn {
    public void close() throws IOException {
        try {
            RequestHeader h = new RequestHeader();
            h.setType(ZooDefs.OpCode.closeSession);
            submitRequest(h, null, null, null);
        } catch (InterruptedException e) {
            // ignore, close the send/event threads
        } finally {
            // 关闭连接
            disconnect();
        }
    }
    
    public void disconnect() {
        // SendThread发起关闭
        sendThread.close();
        eventThread.queueEventOfDeath();
    }
}
class SendThread extends ZooKeeperThread {
    void close() {
        // 状态设置为CLOSED
        state = States.CLOSED;
        clientCnxnSocket.wakeupCnxn();
    }
}

总结：

我们需要注意的是，在使用Zookeeper客户端连接服务端时，若服务端发生异常，导致连接中断时，则客户端会不断发起重连，不断的重试Zookeeper集群中的其他机器，重试代码在SendThread.run()方法中。

当客户端在超时时间sessionTimeout还没有连接上，那么服务端会清理该会话，此时客户端哪怕连接上，也会被服务端认为是非法会话而中断。这时我们则只能手动新建一个Zookeeper客户端会话。

最后，借用一张网络上图片来总结下会话状态的转换（来自【分布式】Zookeeper会话 - leesf - 博客园 ）