zk源码—4.会话的实现原理一（会话层的基本功能是什么）

大纲

1.创建会话

(1)客户端的会话状态

(2)服务端的会话创建

(3)会话ID的初始化实现

(4)设置的会话超时时间没生效的原因

2.分桶策略和会话管理

(1)分桶策略和过期队列

(2)会话激活

(3)会话超时检查

(4)会话清理

1.创建会话

(1)客户端的会话状态

(2)服务端的会话创建

(3)会话ID的初始化实现

(4)设置的会话超时时间没生效的原因

会话是zk中最核心的概念之一，客户端与服务端的交互都离不开会话的相关操作。其中包括临时节点的生命周期、客户端请求的顺序、Watcher通知机制等。比如会话关闭时，服务端会自动删除该会话所创建的临时节点。当客户端会话退出，通过Watcher机制可向订阅该事件的客户端发送通知。

(1)客户端的会话状态

当zk客户端与服务端成功建立连接后，就会创建一个会话。在zk客户端的运行过程(会话生命周期)中，会话会经历不同的状态变化。

这些不同的会话状态包括：正在连接(CONNECTING)、已经连接(CONNECTED)、会话关闭(CLOSE)、正在重新连接(RECONNECTING)、已经重新连接(RECONNECTED)等。

如果zk客户端需要与服务端建立连接创建一个会话，那么客户端就必须提供一个使用字符串表示的zk服务端地址列表。

当客户端刚开始创建ZooKeeper对象时，其会话状态就是CONNECTING，之后客户端会根据服务端地址列表中的IP地址分别尝试进行网络连接。如果成功连接上zk服务端，那么客户端的会话状态就会变为CONNECTED。

如果因为网络闪断或者其他原因造成客户端与服务端之间的连接断开，那么zk客户端会自动进行重连操作，同时其会话状态变为CONNECTING，直到重新连接上zk服务端后，客户端的会话状态才变回CONNECTED。

通常 总是在CONNECTING或CONNECTED间切换。如果出现会话超时、权限检查失败、客户端主动退出程序等情况，那么客户端的会话状态就会直接变为CLOSE。

public class CreateSessionDemo {
    private final static String CONNECTSTRING = "192.168.1.5:2181";
    private static CountDownLatch countDownLatch = new CountDownLatch(1);
    
    public static void main(String[] args) throws Exception {
        //创建zk
        ZooKeeper zooKeeper = new ZooKeeper(CONNECTSTRING, 5000, new Watcher() {
            public void process(WatchedEvent watchedEvent) {
                //如果当前的连接状态是连接成功, 则通过计数器去控制, 否则进行阻塞, 因为连接是需要时间的
                //如果已经获得连接了, 那么状态会是SyncConnected
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected){
                    countDownLatch.countDown();
                    System.out.println(watchedEvent.getState());
                }
                //如果数据发生了变化
                if (watchedEvent.getType() == Event.EventType.NodeDataChanged) {
                    System.out.println("节点发生了变化, 路径: " + watchedEvent.getPath());
                }
            }
        });
        //进行阻塞
        countDownLatch.await();
        //确定已经获得连接了再进行zk的操作: 增删改查
        ...
    }
}

public class ZooKeeper implements AutoCloseable {
    protected final ClientCnxn cnxn;
    protected final ZKWatchManager watchManager;//ZKWatchManager实现了ClientWatchManager
    ...
    //1.初始化ZooKeeper对象
    public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly, HostProvider aHostProvider, ZKClientConfig clientConfig) throws IOException {
        ...
        //创建客户端的Watcher管理器ZKWatchManager
        watchManager = defaultWatchManager();
        //2.设置会话默认的Watcher，保存在客户端的Watcher管理器ZKWatchManager中
        watchManager.defaultWatcher = watcher;
        ConnectStringParser connectStringParser = new ConnectStringParser(connectString);
        //3.构造服务器地址列表管理器StaticHostProvider
        hostProvider = aHostProvider;
        //4.创建并初始化客户端的网络连接器ClientCnxn + 5.初始化SendThread和EventThread
        cnxn = createConnection(connectStringParser.getChrootPath(), hostProvider, sessionTimeout, this, watchManager, getClientCnxnSocket(), canBeReadOnly);
        //6.启动SendThread和EventThread
        cnxn.start();
    }
    
    protected ClientCnxn createConnection(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper, ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket, boolean canBeReadOnly) throws IOException { 
        return new ClientCnxn(chrootPath, hostProvider, sessionTimeout, this, watchManager, clientCnxnSocket, canBeReadOnly);
    }
    
    //从配置中获取客户端使用的网络连接配置：使用NIO还是Netty，然后通过反射进行实例化客户端Socket
    private ClientCnxnSocket getClientCnxnSocket() throws IOException {
        String clientCnxnSocketName = getClientConfig().getProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET);
        if (clientCnxnSocketName == null) {
            clientCnxnSocketName = ClientCnxnSocketNIO.class.getName();
        }
        Constructor<?> clientCxnConstructor = Class.forName(clientCnxnSocketName).getDeclaredConstructor(ZKClientConfig.class);
        ClientCnxnSocket clientCxnSocket = (ClientCnxnSocket) clientCxnConstructor.newInstance(getClientConfig());
        return clientCxnSocket;
    }
    
    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;
        }
    }
    ...
    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>>();
        protected volatile Watcher defaultWatcher;
        ...
    }
    
    protected ZKWatchManager defaultWatchManager() {
        //创建客户端的Watcher管理器ZKWatchManager
        return new ZKWatchManager(getClientConfig().getBoolean(ZKClientConfig.DISABLE_AUTO_WATCH_RESET));
    }
    ...
}

public class ClientCnxn {
    ...
    volatile States state = States.NOT_CONNECTED;
    private final HostProvider hostProvider;
    
    public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper, ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket, long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) {
        ...
        this.hostProvider = hostProvider;
        //5.初始化SendThread和EventThread
        sendThread = new SendThread(clientCnxnSocket);
        eventThread = new EventThread();
        ...
    }
    
    //6.启动SendThread和EventThread
    public void start() {
        sendThread.start();
        eventThread.start();
    }
    
    class SendThread extends ZooKeeperThread {
        private final ClientCnxnSocket clientCnxnSocket;
        ...
        SendThread(ClientCnxnSocket clientCnxnSocket) {
            super(makeThreadName("-SendThread()"));
            //客户端刚开始创建ZooKeeper对象时，设置其会话状态为CONNECTING
            state = States.CONNECTING;
            this.clientCnxnSocket = clientCnxnSocket;
            //设置为守护线程
            setDaemon(true);
        }
        
        @Override
        public void run() {
            ...
            while (state.isAlive()) {
                ...
                //7.获取其中一个zk服务端的地址
                serverAddress = hostProvider.next(1000);
                //向zk服务端发起连接请求
                startConnect(serverAddress);
                ...
            }
            ...
        }
        
        private void startConnect(InetSocketAddress addr) throws IOException {
            ...
            state = States.CONNECTING;
            //8.创建TCP连接
            //接下来以ClientCnxnSocketNetty的connect为例
            clientCnxnSocket.connect(addr);
        }
        
        void onConnected(int _negotiatedSessionTimeout, long _sessionId, byte[] _sessionPasswd, boolean isRO) throws IOException {
            ...
            //和服务端建立连接后的处理
            state = (isRO) ? States.CONNECTEDREADONLY : States.CONNECTED;
            ...
        }
        ...
    }
}

public class ClientCnxnSocketNetty extends ClientCnxnSocket {
    //向zk服务端发起建立连接的请求
    @Override
    void connect(InetSocketAddress addr) throws IOException {
        ...
        Bootstrap bootstrap = new Bootstrap().group(eventLoopGroup)
            .channel(NettyUtils.nioOrEpollSocketChannel())
            .option(ChannelOption.SO_LINGER, -1).option(ChannelOption.TCP_NODELAY, true)
            .handler(new ZKClientPipelineFactory(addr.getHostString(), addr.getPort()));
        bootstrap = configureBootstrapAllocator(bootstrap);
        bootstrap.validate();
        connectFuture = bootstrap.connect(addr);
        ...
    }
    
    private class ZKClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
        ...
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) throws Exception {
            ...
            //与zk服务端建立好连接后的处理，调用父类ClientCnxnSocket的readConnectResult()方法
            readConnectResult();
            ...
        }
        ...
    }
    ...
}

abstract class ClientCnxnSocket {
    void readConnectResult() throws IOException {
        ...
        sendThread.onConnected(conRsp.getTimeOut(), this.sessionId, conRsp.getPasswd(), isRO);
    }
    ...
}

(2)服务端的会话创建

在zk服务端中，使用SessionImpl表示客户端与服务器端连接的会话实体。SessionImpl由三个部分组成：会话ID(sessionID)、会话超时时间(timeout)、会话关闭状态(isClosing)。

一.会话ID

会话ID是一个会话的标识符，当创建一次会话时，zk服务端会自动为其分配一个唯一的ID。

二.会话超时时间

一个会话的超时时间就是指一次会话从发起后到被服务器关闭的时长。设置会话超时时间后，zk服务端会参考设置的超时时间，最终计算一个服务端自己的超时时间。这个超时时间才是真正被zk服务端用于管理用户会话的超时时间。

三.会话关闭状态

会话关闭状态isClosing表示一个会话是否已经关闭。如果zk服务端检查到一个会话已经因为超时等原因失效时，就会将该会话的isClosing标记为关闭，之后就不再对该会话进行操作。

public class SessionTrackerImpl extends ZooKeeperCriticalThread implements SessionTracker {
    ...
    public static class SessionImpl implements Session {
        SessionImpl(long sessionId, int timeout) {
            this.sessionId = sessionId;
            this.timeout = timeout;
            isClosing = false;
        }

        final long sessionId;//会话ID
        final int timeout;//会话超时时间
        boolean isClosing;//会话关闭状态
        ...
    }
    ...
}

服务端收到客户端的创建会话请求后，进行会话创建的过程大概分四步：处理ConnectRequest请求、创建会话、请求处理链处理和会话响应。

步骤一：处理ConnectRequest请求

首先由NettyServerCnxn负责接收来自客户端的创建会话请求，然后反序列化出ConnectRequest对象，并完成会话超时时间的协商。

步骤二：创建会话

SessionTrackerImpl的createSession()方法会为该会话分配一个sessionID，并将该sessionID注册到sessionsById和sessionsWithTimeout中，同时通过SessionTrackerImpl的updateSessionExpiry()方法进行会话激活。

步骤三：请求处理链处理

接着调用
ZooKeeperServer.firstProcessor的processRequest()方法，让该会话请求会在zk服务端的各个请求处理器之间进行顺序流转。

步骤四：会话响应

最后在请求处理器FinalRequestProcessor的processRequest()方法中进行会话响应。

//由网络连接工厂类监听到客户端的创建会话请求
public class NettyServerCnxnFactory extends ServerCnxnFactory {
    class CnxnChannelHandler extends ChannelDuplexHandler {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ...
            NettyServerCnxn cnxn = ctx.channel().attr(CONNECTION_ATTRIBUTE).get();
            cnxn.processMessage((ByteBuf) msg);
            ...
        }
        ...
    }
    ...
}

public class NettyServerCnxn extends ServerCnxn {
    private volatile ZooKeeperServer zkServer;
    
    void processMessage(ByteBuf buf) {
        ...
        receiveMessage(buf);
        ...
    }
    
    private void receiveMessage(ByteBuf message) {
        ...
        ZooKeeperServer zks = this.zkServer;
        //处理会话连接请求
        zks.processConnectRequest(this, bb);
        ...
    }
    ...
}

public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
    protected SessionTracker sessionTracker;
    ...
    public synchronized void startup() {
        startupWithServerState(State.RUNNING);
    }
    
    private void startupWithServerState(State state) {
        //创建并启动会话管理器
        if (sessionTracker == null) {
            createSessionTracker();
        }
        startSessionTracker();
        //初始化请求处理链
        setupRequestProcessors();
        ...
    }
    
    protected void createSessionTracker() {
        sessionTracker = new SessionTrackerImpl(this, zkDb.getSessionWithTimeOuts(), tickTime, createSessionTrackerServerId, getZooKeeperServerListener());
    }
    
    public void processConnectRequest(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException {
        //步骤一：处理ConnectRequest请求
        BinaryInputArchive bia = BinaryInputArchive.getArchive(new ByteBufferInputStream(incomingBuffer));
        ConnectRequest connReq = new ConnectRequest();
        connReq.deserialize(bia, "connect");
        ...
        //协商会话超时时间
        int sessionTimeout = connReq.getTimeOut();
        byte passwd[] = connReq.getPasswd();
        int minSessionTimeout = getMinSessionTimeout();
        if (sessionTimeout < minSessionTimeout) {
            sessionTimeout = minSessionTimeout;
        }
        int maxSessionTimeout = getMaxSessionTimeout();
        if (sessionTimeout > maxSessionTimeout) {
            sessionTimeout = maxSessionTimeout;
        }
        cnxn.setSessionTimeout(sessionTimeout);
        ...
        long sessionId = connReq.getSessionId();
        if (sessionId == 0) {
            //步骤二：创建会话
            long id = createSession(cnxn, passwd, sessionTimeout);
        } else {
            long clientSessionId = connReq.getSessionId();
            if (serverCnxnFactory != null) {
                serverCnxnFactory.closeSession(sessionId);
            }
            if (secureServerCnxnFactory != null) {
                secureServerCnxnFactory.closeSession(sessionId);
            }
            cnxn.setSessionId(sessionId);
            reopenSession(cnxn, sessionId, passwd, sessionTimeout);
        }
    }
    
    long createSession(ServerCnxn cnxn, byte passwd[], int timeout) {
        if (passwd == null) {
            passwd = new byte[0];
        }
        //通过会话管理器创建会话
        long sessionId = sessionTracker.createSession(timeout);
        Random r = new Random(sessionId ^ superSecret);
        r.nextBytes(passwd);
        ByteBuffer to = ByteBuffer.allocate(4);
        to.putInt(timeout);
        cnxn.setSessionId(sessionId);
        Request si = new Request(cnxn, sessionId, 0, OpCode.createSession, to, null);
        setLocalSessionFlag(si);
        //激活会话 + 提交请求到请求处理链进行处理
        submitRequest(si);
        return sessionId;
    }
    
    public void submitRequest(Request si) {
        ...
        //激活会话
        touch(si.cnxn);
        //步骤三：交给请求处理链进行处理，在FinalRequestProcessor中会进行会话响应
        firstProcessor.processRequest(si);
        ...
    }
    ...
}

public class SessionTrackerImpl extends ZooKeeperCriticalThread implements SessionTracker {
    ...
    private final AtomicLong nextSessionId = new AtomicLong();
    private final ExpiryQueue<SessionImpl> sessionExpiryQueue;
    
    public SessionTrackerImpl(SessionExpirer expirer, ConcurrentMap<Long, Integer> sessionsWithTimeout, 
            int tickTime, long serverId, ZooKeeperServerListener listener) {
        super("SessionTracker", listener);
        this.expirer = expirer;
        this.sessionExpiryQueue = new ExpiryQueue<SessionImpl>(tickTime);
        this.sessionsWithTimeout = sessionsWithTimeout;
        //初始化SessionId
        this.nextSessionId.set(initializeNextSession(serverId));
        for (Entry<Long, Integer> e : sessionsWithTimeout.entrySet()) {
            addSession(e.getKey(), e.getValue());
        }
        EphemeralType.validateServerId(serverId);
    }
    ...
    public long createSession(int sessionTimeout) {
        //为会话分配一个sessionID
        long sessionId = nextSessionId.getAndIncrement();
        //将sessionID注册到sessionsById和sessionsWithTimeout中
        addSession(sessionId, sessionTimeout);
        return sessionId;
    }
    
    public synchronized boolean addSession(long id, int sessionTimeout) {
        sessionsWithTimeout.put(id, sessionTimeout);
        boolean added = false;
        SessionImpl session = sessionsById.get(id);
        if (session == null) {
            session = new SessionImpl(id, sessionTimeout);
        }

        SessionImpl existedSession = sessionsById.putIfAbsent(id, session);
        if (existedSession != null) {
            session = existedSession;
        } else {
            added = true;
            LOG.debug("Adding session 0x" + Long.toHexString(id));
        }
        ...
        updateSessionExpiry(session, sessionTimeout);
        return added;
    }
    
    private void updateSessionExpiry(SessionImpl s, int timeout) {
        ...
        sessionExpiryQueue.update(s, timeout);
    }
    ...
}

public class FinalRequestProcessor implements RequestProcessor {
    ...
    public void processRequest(Request request) {
        ...
        ServerCnxn cnxn = request.cnxn;
        //步骤四：会话响应
        cnxn.sendResponse(hdr, rsp, "response");
        ...
    }
    ...
}

public abstract class ServerCnxn implements Stats, Watcher {
    ...
    public void sendResponse(ReplyHeader h, Record r, String tag) throws IOException {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
        try {
            baos.write(fourBytes);
            bos.writeRecord(h, "header");
            if (r != null) {
                bos.writeRecord(r, tag);
            }
            baos.close();
        } catch (IOException e) {
            LOG.error("Error serializing response");
        }
        byte b[] = baos.toByteArray();
        serverStats().updateClientResponseSize(b.length - 4);
        ByteBuffer bb = ByteBuffer.wrap(b);
        bb.putInt(b.length - 4).rewind();
        sendBuffer(bb);
    }
    ...
}

public class NettyServerCnxn extends ServerCnxn {
    ...
    @Override
    public void sendBuffer(ByteBuffer sendBuffer) {
        if (sendBuffer == ServerCnxnFactory.closeConn) {
            close();
            return;
        }
        channel.writeAndFlush(Unpooled.wrappedBuffer(sendBuffer)).addListener(onSendBufferDoneListener);
    }
    ...
}

(3)会话ID的初始化实现

SessionTracker是zk服务端的会话管理器，zk会话的整个生命周期都离不开SessionTracker的参与。SessionTracker是一个接口类型，规定了会话管理的相关操作行为，具体的会话管理逻辑则由SessionTrackerImpl来完成。

SessionTrackerImpl类实现了SessionTracker接口，其中有四个关键字段：sessionExpiryQueue字段表示的是会话过期队列，用于管理会话自动过期。nextSessionId字段记录了当前生成的会话ID。sessionsById字段用于根据会话ID来管理具体的会话实体。sessionsWithTimeout字段用于根据会话ID管理会话的超时时间。

在SessionTrackerImpl类初始化时，会调用initializeNextSession()方法来生成一个初始化的会话ID。之后在zk的运行过程中，会在该会话ID的基础上为每个会话分配ID。

public class SessionTrackerImpl extends ZooKeeperCriticalThread implements SessionTracker {
    ...
    private final ExpiryQueue<SessionImpl> sessionExpiryQueue;//过期队列
    private final AtomicLong nextSessionId = new AtomicLong();//当前生成的会话ID
    ConcurrentHashMap<Long, SessionImpl> sessionsById;//根据会话ID来管理具体的会话实体
    ConcurrentMap<Long, Integer> sessionsWithTimeout;//根据不同的会话ID管理每个会话的超时时间
    
    public SessionTrackerImpl(SessionExpirer expirer, ConcurrentMap<Long, Integer> sessionsWithTimeout, 
            int tickTime, long serverId, ZooKeeperServerListener listener) {
        super("SessionTracker", listener);
        this.expirer = expirer;
        this.sessionExpiryQueue = new ExpiryQueue<SessionImpl>(tickTime);
        this.sessionsWithTimeout = sessionsWithTimeout;
        //初始化SessionId
        this.nextSessionId.set(initializeNextSession(serverId));
        for (Entry<Long, Integer> e : sessionsWithTimeout.entrySet()) {
            addSession(e.getKey(), e.getValue());
        }
        EphemeralType.validateServerId(serverId);
    }
    
    public static long initializeNextSession(long id) {
        long nextSid;
        nextSid = (Time.currentElapsedTime() << 24) >>> 8;
        nextSid = nextSid | (id <<56);
        return nextSid;
    }
    ...
}

在SessionTrackerImpl的initializeNextSession()方法中，生成初始化的会话ID的过程如下：

步骤一：获取当前时间的毫秒表示

步骤二：将得到的毫秒表示的时间先左移24位

步骤三：将左移24位后的结果再右移8位

步骤四：服务器SID左移56位

步骤五：将右移8位的结果和左移56位的结果进行位与运算

算法概述：高8位确定所在机器，低56位使用当前时间的毫秒表示来进行随机。

其中左移24位的目的是：将毫秒表示的时间的最高位的1移出，可以防止出现负数。

//时间的二进制表示，41位
10100000110000011110001000100110111110111
//左移24位变成，64位
0100000110000011110001000100110111110111000000000000000000000000
//右移8位变成，64位
0000000001000001100000111100010001001101111101110000000000000000
//假设服务器SID为2，那么左移56位变成
0000001000000000000000000000000000000000000000000000000000000000
//位与运算
0000001001000001100000111100010001001101111101110000000000000000

(4)设置的会话超时时间没生效的原因

在平时的开发工作中，最常遇到的场景就是会话超时异常。zk的会话超时异常包括：客户端readTimeout异常和服务端sessionTimeout异常。

需要注意的是：可能虽然设置了超时时间，但实际服务运行时zk并没有按设置的超时时间来管理会话。

这是因为实际起作用的超时时间是由客户端和服务端协商决定的。zk客户端在和服务端建立连接时，会提交一个客户端设置的会话超时时间，而该超时时间会和服务端设置的最大超时时间和最小超时时间进行比较。如果正好在服务端设置允许的范围内，则采用客户端的超时时间管理会话。如果大于或小于服务端设置的超时时间，则采用服务端的超时时间管理会话。