暂无图片
暂无图片
暂无图片
暂无图片
暂无图片
首页 / 聊聊Zookeeper的会话(中篇)

聊聊Zookeeper的会话(中篇)

糖爸的架构师之路 2021-06-24
1235

写在前面

在上一篇文章聊聊Zookeeper的会话(上)中，我们介绍了Zookeeper的会话在客户端的创建流程。我们知道，一次完整会话的建立分为客户端会话创建和服务端会话创建。而且在Zookeeper中，会话的管理是由服务端来完成的。所以我们本文就来介绍一下服务端侧是如何管理会话的（基于Zookeeper 3.6.1版本）。服务端的会话管理主要涉及到下面几个类

  • Session、SessionImpl

  • SessionTracker、SessionTrackerImpl

服务端的会话管理涉及到以下几个部分：

  • 会话激活

  • 会话分桶

  • 会话超时检查

  • 会话清理

下面我们来逐一看一下
Session
要想了解服务端侧是如何管理会话的，需要先了解一下Session在Zookeeper中的定义。Session是ZooKeeper中的会话实体，代表了一个客户端会话。其权限定类名为
org.apache.zookeeper.server.SessionTracker.Session。Session本身是一个接口，实现类为org.apache.zookeeper.server.SessionTrackerImpl.SessionImpl，具体包含以下3个基本属性

  • sessionID：会话 ID，用来唯一标识一个会话，每次客户端创建新会话的时候，ZooKeeper都会为其分配一个全局唯一的 sessionID。

  • TimeOut：会话超时时间。客户端在构造 ZooKeeper 实例的时候，会配置一个sessionTimeout参数用于指定会话的超时时间。ZooKeeper客户端向服务器发送这个超时时间后，服务器会根据自己的超时时间限制最终确定会话的超时时间。

  • isClosing：该属性用于标记一个会话是否已经被关闭。通常当服务端检测到一个会话已经超时失效的时候，会将该会话的isClosing 属性标记为"已关闭"，这样就能确保不再处理来自该会话的新请求了。

下面是具体代码：


SessionId

上一篇文章聊聊Zookeeper的会话(上)中我们介绍过SessionId，现在我们再来回顾一下。SessionId是用来唯一标识一个会话。当客户端发起会话创建请求后由服务端生成并下发给客户端。因此ZooKeeper必须保证SessionId的全局唯一性。在每次客户端向服务端发起"会话创建"请求时，服务端都会为其分配一个SessionId。那么SessionId是如何在服务端侧生成的呢？我们来看一下具体的代码：

在 SessionTracker初始化的时候，会调用 initializeNextSession()方法来生成一个初始化的 SessionId，之后在ZooKeeper的正常运行过程中，会在该 SessionId的基础上为每个会话进行分配。初始化步骤如下：

从上面的初始化方法中可以看出，该方法的入参是一个Long类型的整数id，该id表示的是服务端的机器编号，这个参数是在部署ZooKeeper服务器的时候，配置在myid文件中的。

  1. 生成系统当前时间的时间戳，64位的long型整数

  2. 将时间戳左移24位，在无符号右移8位

  3. 经过上一步，该时间戳的高8位全部为0，低56位不为0

  4. 接着，将机器编号左移56位，那么机器编号的高8位不为0，低56位全为0

  5. 最后，将上面得到的机器编号和时间戳进行或运算


为什么要做这样的位运算？

         时间戳经过这样的运算之后，高8位全部为0，和机器编号的高8位进行或运算之后，其结果完全取决于机器编号的高8位；同理，低56位由时间戳决定。因此可以看出，SessionId其实是由机器编号+时间戳唯一决定的。可以保证在单机环境下的唯一性。


为什么右移8位需要采用无符号？

之所以右移8位的时候采用无符号，是因为防止前面左移24位的时候，可能出现负数的情况，因此为了消除产生的负数的影响，采用无符号的右移。

SessionTracker
SessionTracker是ZooKeeper服务端的会话管理器，负责会话的创建、管理和清理等工作。可以说，整个会话的生命周期都离不开 SessionTracker 的管理。SessionTracker本身是一个接口，它的实现类为SessionTrackerImpl，其内部结构如下：

  • sessionsById：ConcurrentHashMap<Long，SessionImpl>类型，用于根据 sessionID来管理 Session 实体。

  • sessionsWithTimeout：ConcurrentHashMap<Long，Integer>类型，用于根据SessionId来管理会话的超时时间。该数据结构和ZooKeeper 内存数据库相连通，会被定期持久化到快照文件中去。

  • sessionExpiryQueue：过期队列。用于维护会话的过期，并且使用bucket来维护会话，每一个bucket对应一个某时间范围内过期的会话。

那么SessionTracker是如何管理会话的呢？下面我们来详细看一看


会话管理

ZooKeeper 的会话管理主要是由SessionTracker 负责的，其采用了一种特殊的会话管理方式，我们称之为"分桶策略"。


分桶策略

所谓分桶策略，是指将类似的会话放在同一区块中进行管理，以便于 Zookeeper 对会话进行不同区块的隔离处理以及同一区块的统一处理。分桶策略使用到的类为ExpireQueue。上面我们提到过，在SessionTrackerImpl中维护了一个过期队列sessionExpiryQueue，它的类型即是ExpiryQueue。我们首先来看一下ExpiryQueue内部的结构：
    // 保存Session对象与其过期时间额映射关系，key是Session，value过期时间
    private final ConcurrentHashMap<E, Long> elemMap = new ConcurrentHashMap<E, Long>();
    // 保存过期时间与在此过期时间点要过期的会话集合的映射关系。
    // key是过期时间，value是Session集合（会话桶）
    private final ConcurrentHashMap<Long, Set<E>> expiryMap = new ConcurrentHashMap<Long, Set<E>>();
    // 下一个过期的时间点
    private final AtomicLong nextExpirationTime = new AtomicLong();
    // 过期时间间隔
    private final int expirationInterval;

    下图是分桶策略的示意图：

    ZooKeeper将所有的会话都分配在了不同的区块之中，分配的原则是每个会话的"下次超时时间点"（ExpirationTime）。ExpirationTime是指该会话最近一次可能超时的时间点，对于一个新创建的会话而言，其会话创建完毕后，ZooKeeper就会为其计算 ExpirationTime，计算方式如下∶
      ExpirationTime = CurrentTime + SessionTimeout

      • CurrentTime：当前时间，单位是毫秒

      • SessionTimeout：该会话设置的超时时间，单位也是毫秒。

      那么，上图中横坐标所标识的时间，是否就是通过上述公式计算出来的呢？答案是否定的，在ZooKeeper的实际实现中，还做了一个处理。ZooKeeper的Leader服务器在运行期间会定时地进行会话超时检查，其时间间隔是 Expirationlnterval，单位是毫秒，默认值是 tickTime的值。即默认情况下，每隔 2000毫秒进行一次会话超时检查。为了方便对多个会话同时进行超时检查，完整的ExpirationTime的计算方式如下∶

      这里参数time就是之前计算的ExpirationTime。所以综合上面两个计算，可以得出ExpirationTime的最终计算逻辑为：
        ExpirationTime_ = CurrentTime + SessionTimeout
        ExpirationTime = (ExpirationTime_ / ExpirationInterval + 1) * ExpirationInterval
        这里可能小伙伴看的比较绕。为了便于理解，我们可以举几个例子，假设zk默认的时间间隔为2000ms：

        • 当sessionA在1500ms后过期时，那么其会坐落在(1500/2000+1)*2000=2000ms这个key里。

        • 当sessionB在3000ms后过期时，那么其会坐落在(3000/2000+1)*2000=4000ms这个key里。

        • 当sessionC在5000ms后过期时，那么其会坐落在(5000/2000+1)*2000=6000ms这个key里。

        • 当sessionD在7500ms后过期时，那么其会坐落在(7500/2000+1)*2000=8000ms这个key里。

        0
        		2000ms4000ms6000ms
        		8000ms

        		SessionASessionBSessionCSessionD
        这样处理过之后，就可以将离散的Session过期时间根据一定的规则归档到不同的key中，线程就不用遍历所有的会话去注意检查他们的过期时间了。类似于我们在整理衣服时，将不同季节的衣服放到不同的格子中分门别类，这样在换季时就不需要每件衣服都拿出来看一下适合什么季节，而是直接找到当前季节对应的格子，在对应的格子中找到合适的衣服就可以。

        会话激活

        从上面看来，session似乎是到了事先计算好的时间就会过期，其实并非如此。为了保持客户端会话的有效性，在ZooKeeper的运行过程中，客户端会在会话超时时间过期范围内向服务端发送 PING 请求来保持会话的有效性，我们俗称"心跳检测"。同时，服务端需要不断地接收来自客户端的这个心跳检测，并且需要重新激活对应的客户端会话，我们将这个重新激活的过程称为TouchSession。会话激活的过程，不仅能够使服务端检测到对应客户端的存活性，同时也能让客户端自己保持连接状态。我们简单讲一下流程：

        1. 检查该会话是否被关闭。如果关闭，则不再激活。

        2. 计算新的超时时间

        3. 迁移会话（从老桶到新桶）

        具体方法在SessionTrackerImpl.touch()中，我们来看一下具体的代码：

        updateSessionExpiry()方法内部会调用ExpireQueue.update()根据分桶策略将会话放到指定的会话桶中，如果在旧桶中存在当前会话，需要将该会话移动到新桶中，并将旧桶中的该会话删掉。


        超时检查
        在Zookeeper中，会话超时检查同样是由SessionTracker 负责的。
        SessionTracker中有一个单独的线程专门进行会话超时检查，这里我们将其称为"超时检查线程"，其工作机制的核心思路其实非常简单∶逐个依次地对会话桶中剩下的会话进行清理。如果一个会话被激活，那么ZooKeeper 会将其从上一个会话桶迁移到下一个会话桶中。如果触发了会话激活，ZooKeeper 会将其从 expirationTime 1桶迁移到 expirationTime n桶中去。于是，expirationTime 1中留下的所有会话都是尚未被激活的。因此，超时检查线程的任务就是定时检查出这个会话桶中所有剩下的未被迁移的会话。
        下面我们来看一下具体的代码：
        在超时线程中，会通过while循环不断的处理将要过期的会话


        获取距离最近一次过期时间时长方法，获取到时长后执行此时长的线程等待(Thread.sleep())

        在ExpireQueue.poll()方法中，首先更新最近一次的过期时间，从expiryMap中移除要过期的会话桶，并将此会话桶返回进行过期处理。


        最后附上服务端会话管理的整体流程(图片来自网络)~

        以上就是Zookeeper在服务端侧针对会话管理的完整处理过程。至于客户端发送请求后到服务端是如何创建会话的，我们放到下一章节来介绍。

        敬请期待吧~~

        数据库
        文章转载自糖爸的架构师之路，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

        评论