掘金 后端 ( ) • 2021-06-22 16:00
深入浅出Zookeeper源码（三）：会话管理
session 源码 param session对象
.markdown-body{word-break:break-word;line-height:1.75;font-weight:400;font-size:15px;overflow-x:hidden;color:#333}.markdown-body h1,.markdown-body h2,.markdown-body h3,.markdown-body h4,.markdown-body h5,.markdown-body h6{line-height:1.5;margin-top:35px;margin-bottom:10px;padding-bottom:5px}.markdown-body h1{font-size:30px;margin-bottom:5px}.markdown-body h2{padding-bottom:12px;font-size:24px;border-bottom:1px solid #ececec}.markdown-body h3{font-size:18px;padding-bottom:0}.markdown-body h4{font-size:16px}.markdown-body h5{font-size:15px}.markdown-body h6{margin-top:5px}.markdown-body p{line-height:inherit;margin-top:22px;margin-bottom:22px}.markdown-body img{max-width:100%}.markdown-body hr{border:none;border-top:1px solid #ddd;margin-top:32px;margin-bottom:32px}.markdown-body code{word-break:break-word;border-radius:2px;overflow-x:auto;background-color:#fff5f5;color:#ff502c;font-size:.87em;padding:.065em .4em}.markdown-body code,.markdown-body pre{font-family:Menlo,Monaco,Consolas,Courier New,monospace}.markdown-body pre{overflow:auto;position:relative;line-height:1.75}.markdown-body pre>code{font-size:12px;padding:15px 12px;margin:0;word-break:normal;display:block;overflow-x:auto;color:#333;background:#f8f8f8}.markdown-body a{text-decoration:none;color:#0269c8;border-bottom:1px solid #d1e9ff}.markdown-body a:active,.markdown-body a:hover{color:#275b8c}.markdown-body table{display:inline-block!important;font-size:12px;width:auto;max-width:100%;overflow:auto;border:1px solid #f6f6f6}.markdown-body thead{background:#f6f6f6;color:#000;text-align:left}.markdown-body tr:nth-child(2n){background-color:#fcfcfc}.markdown-body td,.markdown-body th{padding:12px 7px;line-height:24px}.markdown-body td{min-width:120px}.markdown-body blockquote{color:#666;padding:1px 23px;margin:22px 0;border-left:4px solid #cbcbcb;background-color:#f8f8f8}.markdown-body blockquote:after{display:block;content:""}.markdown-body blockquote>p{margin:10px 0}.markdown-body ol,.markdown-body ul{padding-left:28px}.markdown-body ol li,.markdown-body ul li{margin-bottom:0;list-style:inherit}.markdown-body ol li .task-list-item,.markdown-body ul li .task-list-item{list-style:none}.markdown-body ol li .task-list-item ol,.markdown-body ol li .task-list-item ul,.markdown-body ul li .task-list-item ol,.markdown-body ul li .task-list-item ul{margin-top:0}.markdown-body ol ol,.markdown-body ol ul,.markdown-body ul ol,.markdown-body ul ul{margin-top:3px}.markdown-body ol li{padding-left:6px}.markdown-body .contains-task-list{padding-left:0}.markdown-body .task-list-item{list-style:none}@media (max-width:720px){.markdown-body h1{font-size:24px}.markdown-body h2{font-size:20px}.markdown-body h3{font-size:18px}}

本文首发于泊浮目的简书:www.jianshu.com/u/204b8aaab…

版本日期备注1.02020.3.29文章首发1.12020.4.18改进小结部分1.22020.5.4修复笔误部分1.42020.7.21段落重新排版，增强语义1.52020.8.6增加题图1.62021.6.22标题从深入浅出Zookeeper（三）：会话管理变更为深入浅出Zookeeper源码（三）：会话管理

前言

我们知道zookeeper是一个分布式协同系统。在一个大型的分布式系统中，必然会有大量的client来连接zookeeper。那么zookeeper是如何管理这些session的生命周期呢？带着这个问题，我们进入今天的正文。

Session管理者：SessionTracker

我们先来看看session相关的核心类——位于服务端的SessionTracker的抽象定义：

/**
 * This is the basic interface that ZooKeeperServer uses to track sessions. The
 * standalone and leader ZooKeeperServer use the same SessionTracker. The
 * FollowerZooKeeperServer uses a SessionTracker which is basically a simple
 * shell to track information to be forwarded to the leader.
 */
public interface SessionTracker {
    public static interface Session {
        long getSessionId();
        int getTimeout();
        boolean isClosing();
    }
    public static interface SessionExpirer {
        void expire(Session session);

        long getServerId();
    }

    long createSession(int sessionTimeout);

    /**
     * Add a global session to those being tracked.
     * @param id sessionId
     * @param to sessionTimeout
     * @return whether the session was newly added (if false, already existed)
     */
    boolean addGlobalSession(long id, int to);

    /**
     * Add a session to those being tracked. The session is added as a local
     * session if they are enabled, otherwise as global.
     * @param id sessionId
     * @param to sessionTimeout
     * @return whether the session was newly added (if false, already existed)
     */
    boolean addSession(long id, int to);

    /**
     * @param sessionId
     * @param sessionTimeout
     * @return false if session is no longer active
     */
    boolean touchSession(long sessionId, int sessionTimeout);

    /**
     * Mark that the session is in the process of closing.
     * @param sessionId
     */
    void setSessionClosing(long sessionId);

    /**
     *
     */
    void shutdown();

    /**
     * @param sessionId
     */
    void removeSession(long sessionId);

    /**
     * @param sessionId
     * @return whether or not the SessionTracker is aware of this session
     */
    boolean isTrackingSession(long sessionId);

    /**
     * Checks whether the SessionTracker is aware of this session, the session
     * is still active, and the owner matches. If the owner wasn't previously
     * set, this sets the owner of the session.
     *
     * UnknownSessionException should never been thrown to the client. It is
     * only used internally to deal with possible local session from other
     * machine
     *
     * @param sessionId
     * @param owner
     */
    public void checkSession(long sessionId, Object owner)
            throws KeeperException.SessionExpiredException,
            KeeperException.SessionMovedException,
            KeeperException.UnknownSessionException;

    /**
     * Strictly check that a given session is a global session or not
     * @param sessionId
     * @param owner
     * @throws KeeperException.SessionExpiredException
     * @throws KeeperException.SessionMovedException
     */
    public void checkGlobalSession(long sessionId, Object owner)
            throws KeeperException.SessionExpiredException,
            KeeperException.SessionMovedException;

    void setOwner(long id, Object owner) throws SessionExpiredException;

    /**
     * Text dump of session information, suitable for debugging.
     * @param pwriter the output writer
     */
    void dumpSessions(PrintWriter pwriter);

    /**
     * Returns a mapping of time to session IDs that expire at that time.
     */
    Map> getSessionExpiryMap();
}

复制代码

大致可以看到，该interface定义对会话一系列的控制方法：比如会话的创建、激活及删除等等。

那么我们来看下其SessionTrackerImpl实现中比较重要的接口和成员变量以及方法。

会话的属性与状态

接下来我们来看看一个会话实例会包含哪些属性，话不多说，直接看接口定义：

    public static interface Session {
        long getSessionId();
        int getTimeout();
        boolean isClosing();
    }
复制代码

我们可以看到，在服务端，仅仅记录了client这样的三个属性：sessionId,timeout,isClosing。

但在client，还会更复杂一点。比如session的状态就有好多个：

    @InterfaceAudience.Public
    public enum States {
        CONNECTING, ASSOCIATING, CONNECTED, CONNECTEDREADONLY,
        CLOSED, AUTH_FAILED, NOT_CONNECTED;

        public boolean isAlive() {
            return this != CLOSED && this != AUTH_FAILED;
        }

        /**
         * Returns whether we are connected to a server (which
         * could possibly be read-only, if this client is allowed
         * to go to read-only mode)
         * */
        public boolean isConnected() {
            return this == CONNECTED || this == CONNECTEDREADONLY;
        }
    }

复制代码

通常情况下，因为网络闪断或其他原因，client会出现和server断开的情况。所幸的是，zkClient会自动重连，这时client会变为connecting，直到连上服务器，则变connected。如果会话超时、权限检查失败或client退出程序等异常情况，则客户端会变成close状态。

重要成员变量

    protected final ConcurrentHashMap sessionsById =
        new ConcurrentHashMap();

    private final ExpiryQueue sessionExpiryQueue;

    private final ConcurrentMap sessionsWithTimeout;
复制代码
  • 第一个sessionsById很显然，就是通过session的id与session本体做映射的一个字典。
  • 第二个sessionExpiryQueue，听名字像是一个过期队列，没错，不过里面使用了分桶策略 ，稍后我们会做分析。
  • 第三个sessionsWithTimeout，名字说明一切。用于标示session的超时时间，k是sessionId，v是超时时间。该数据结构和Zk的内存数据库相连通，会被定期持久化到快照里去。

会话管理

会话的创建

要谈会话管理，必然要谈到会话是怎么创建的，不然则显得有些空洞。这里不会赘述client的初始化过程。无论如何，我们需要一个链接，毕竟不能让会话基于空气建立：

  1. 我们的client会随机选一个我们提供的地址，然后委托给ClientCnxnSocket去创建与zk之间的TCP链接。
  2. 接下来SendThread（Client的网络发送线程）构造出一个ConnectRequest请求（代表客户端与服务器创建一个会话）。同时，Zookeeper客户端还会进一步将请求包装成网络IO的Packet对象，放入请求发送队列——outgoingQueue中去。
  3. ClientCnxnSocket从outgoingQueue中取出Packet对象，将其序列化成ByteBuffer后，向服务器进行发送。
  4. 服务端的SessionTracker为该会话分配一个sessionId，并发送响应。
  5. Client收到响应后，此时此刻便明白自己没有初始化，因此会用readConnectResult方法来处理请求。
  6. ClientCnxnSocket会对接受到的服务端响应进行反序列化，得到ConnectResponse对象，并从中获取到Zookeeper服务端分配的会话SessionId。
  7. 通知SendThread，更新Client会话参数（比如重要的connectTimeout），并更新Client状态；另外，通知地址管理器HostProvider当前成功链接的服务器地址。

这就是会话的大致创建流程了，当然我们还省去了SyncConnected-None的事件通知逻辑，因为这在本篇的内容里并不重要。

相关源码：SessionId的分配

    /**
     * Generates an initial sessionId. High order byte is serverId, next 5
     * 5 bytes are from timestamp, and low order 2 bytes are 0s.
     */
    public static long initializeNextSession(long id) {
        long nextSid;
        nextSid = (Time.currentElapsedTime() << 24) >>> 8;
        nextSid =  nextSid | (id <<56);
        if (nextSid == EphemeralType.CONTAINER_EPHEMERAL_OWNER) {
            ++nextSid;  // this is an unlikely edge case, but check it just in case
        }
        return nextSid;
    }
复制代码

简单来说，前7位确定了所在的机器，后57位使用当前时间的毫秒表示进行随机。

会话过期检查

会话过期检查是通过SessionTrackerImpl.run来做的，这是一个线程的核心方法——显然，zk的session过期检查是通过一个线程来做的。

简单来说，ExpiryQueue会根据时间将会要过期的sessions进行归档。比如在12:12:54将会有session1、session2、session3会过期，12:12:55会有session4、session5、session6会过期，那么时间会作为一个k，而对应的过期sessions会被作为一个数组，用字典将它们映射起来：

keyvalue12:12:54[session1,session2,session3]12:12:55[session4,session5,session6]

当然，实际中间隔不会是1s，这里为了便于表达，才这么写的。真实的情况是，zk会计算每个session的过期时间，并将其归档到对应的会话桶中。

  • 计算一个会话的过期时间大致为：CurrentTime+SessionTimeout（见ExpiryQueue的update）。
  • 而归档到Zk的时间节点为：（会话过期时间/ExpirationInterval+1） * ExpirationInterval。

为了便于理解，我们可以举几个例子，Zk默认的间隔时间是2000ms：

  • 比如我们计算出来一个sessionA在3000ms后过期，那么其会坐落在(3000/2000+1)*2000=4000ms这个key里。
  • 比如我们计算出来一个sessionB在1500ms后过期，那么其会坐落在(1500/2000+1)*2000=2000ms这个key里。
02000ms4000ms6000ms8000mssessionBsessionA

这样线程就不用遍历所有的会话去逐一检查它们的过期时间了，有点妙。在这里，也可以简单的讲一下会话清理步骤：

  1. 标记会话为isClosing。这样在会话清理期间接收到客户端的新请求也无法继续处理了。
  2. 发起关闭会话请求给PrepRequestProcessor，使其在整个Zk集群里生效。
  3. 收集需要清理的临时节点 ——在上面提到过sessionsWithTimeout 和内存数据库是共通的。
  4. 发起“节点删除”请求，这个事务会被发到outstandingChanges中去。
  5. 删除临时节点，该逻辑由FinalRequestProcessor触发Zk内存数据库（见FinalRequestProcessor.processRequest）。
  6. 移除会话。从sessionsById 、sessionExpiryQueue 、sessionsWithTimeout 中移除。
  7. 关闭ServerCnxn：从ServerCnxnFactory找出对应的ServerCnxn，将其关闭（见FinalRequestProcessor.closeSession）。

从这里可以了解到，Zk临时节点的自动回收基于会话管理机制。

相关源码：SessionTrackerImpl.run

    @Override
    public void run() {
        try {
            while (running) {
                long waitTime = sessionExpiryQueue.getWaitTime();
                if (waitTime > 0) {
                    Thread.sleep(waitTime);
                    continue;
                }

                for (SessionImpl s : sessionExpiryQueue.poll()) {
                    setSessionClosing(s.sessionId);
                    expirer.expire(s);
                }
            }
        } catch (InterruptedException e) {
            handleException(this.getName(), e);
        }
        LOG.info("SessionTrackerImpl exited loop!");
    }
复制代码

逻辑很简单。去sessionExpiryQueue 里看一下离最近的过期时间还要多久，有的话就等一会儿。

接下来是标记成Closing，并开始做使过期操作。

我们接着看expirer.expire：

  public void expire(Session session) {
        long sessionId = session.getSessionId();
        LOG.info("Expiring session 0x" + Long.toHexString(sessionId)
                + ", timeout of " + session.getTimeout() + "ms exceeded");
        close(sessionId);
    }
复制代码

跳向close:

  private void close(long sessionId) {
        Request si = new Request(null, sessionId, 0, OpCode.closeSession, null, null);
        setLocalSessionFlag(si);
        submitRequest(si);
    }
复制代码

就是build一个新的请求，然后set本地的flag。关键方法是submitRequest:

 public void submitRequest(Request si) {
        if (firstProcessor == null) {
            synchronized (this) {
                try {
                    // Since all requests are passed to the request
                    // processor it should wait for setting up the request
                    // processor chain. The state will be updated to RUNNING
                    // after the setup.
                    while (state == State.INITIAL) {
                        wait(1000);
                    }
                } catch (InterruptedException e) {
                    LOG.warn("Unexpected interruption", e);
                }
                if (firstProcessor == null || state != State.RUNNING) {
                    throw new RuntimeException("Not started");
                }
            }
        }
        try {
            touch(si.cnxn);
            boolean validpacket = Request.isValid(si.type);
            if (validpacket) {
                firstProcessor.processRequest(si);
                if (si.cnxn != null) {
                    incInProcess();
                }
            } else {
                LOG.warn("Received packet at server of unknown type " + si.type);
                new UnimplementedRequestProcessor().processRequest(si);
            }
        } catch (MissingSessionException e) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Dropping request: " + e.getMessage());
            }
        } catch (RequestProcessorException e) {
            LOG.error("Unable to process request:" + e.getMessage(), e);
        }
    }

复制代码

第一段逻辑是等待Processor的chain准备好。接下来是激活一下会话，但会话如果已经被移除或超时，则会抛出异常。这个情况很正常，因为client的session和这里的移除请求并不是同时做的。

接下来则是提交移除会话的请求。

会话激活

从上面看来，session似乎是到了事先计算好的时间就会过期。其实并非如此——client会通过发送请求or心跳请求来保持会话的有效性，即延迟超时时间。这个过程一般叫做touchSession（没错，代码里也是这么叫的）。我们来简单的讲一下流程：

  1. 检查该会话是否被关闭，如果关闭，则不再激活。
  2. 计算新的超时时间（参考上面提到的会话超时计算方法，也可以看ExpiryQueue.update）
  3. 迁移会话（从老桶到新桶）

相关源码：SessionTrackerImpl.touch

    synchronized public boolean touchSession(long sessionId, int timeout) {
        SessionImpl s = sessionsById.get(sessionId);

        if (s == null) {
            logTraceTouchInvalidSession(sessionId, timeout);
            return false;
        }

        if (s.isClosing()) {
            logTraceTouchClosingSession(sessionId, timeout);
            return false;
        }

        updateSessionExpiry(s, timeout);
        return true;
    }
复制代码

获取和校验逻辑不再赘述。直接跳向核心方法ExpiryQueue.update:

    /**
     * Adds or updates expiration time for element in queue, rounding the
     * timeout to the expiry interval bucketed used by this queue.
     * @param elem     element to add/update
     * @param timeout  timout in milliseconds
     * @return         time at which the element is now set to expire if
     *                 changed, or null if unchanged
     */
    public Long update(E elem, int timeout) {
        Long prevExpiryTime = elemMap.get(elem);
        long now = Time.currentElapsedTime();
        Long newExpiryTime = roundToNextInterval(now + timeout);

        if (newExpiryTime.equals(prevExpiryTime)) {
            // No change, so nothing to update
            return null;
        }

        // First add the elem to the new expiry time bucket in expiryMap.
        Set set = expiryMap.get(newExpiryTime);
        if (set == null) {
            // Construct a ConcurrentHashSet using a ConcurrentHashMap
            set = Collections.newSetFromMap(
                new ConcurrentHashMap());
            // Put the new set in the map, but only if another thread
            // hasn't beaten us to it
            Set existingSet = expiryMap.putIfAbsent(newExpiryTime, set);
            if (existingSet != null) {
                set = existingSet;
            }
        }
        set.add(elem);

        // Map the elem to the new expiry time. If a different previous
        // mapping was present, clean up the previous expiry bucket.
        prevExpiryTime = elemMap.put(elem, newExpiryTime);
        if (prevExpiryTime != null && !newExpiryTime.equals(prevExpiryTime)) {
            Set prevSet = expiryMap.get(prevExpiryTime);
            if (prevSet != null) {
                prevSet.remove(elem);
            }
        }
        return newExpiryTime;
    }
复制代码

逻辑非常简单。计算最新的过期时间，并放置到新的归档区间里，再移除掉老归档区间里的会话实例。

小结

在本文中，笔者和大家一起了剖析了zk的session管理机制。有些点我们在以后设计系统时可以借鉴一番：

  • 会话的状态变化主要由client维护，server端保存的状态较少，一定程度上减少了server端的压力。
  • 分桶策略在这种大量client会话场景下显得非常有用，显著提升了会话超时的清理效率。
相关内容
hibernate session管理