详解zookeeper的session管理机制

使用过zookeeper的都知道，当我们使用zookeeper创建一个节点时，我们能选择节点的类型是“临时节点”还是“永久节点”。临时节点和永久节点的区别是，临时节点会在客户端断开连接时被删除，而永久节点无论客户端是否断开连接，都会保留。

临时节点非常重要，我们经常利用它来实现分布式锁、选举等等

而为了实现临时节点的功能，zookeeper服务端就势必要有一套高效的session管理机制，它能实现如下功能：当客户端session失效后，服务端能感知到，随后删掉客户端当前创建的临时节点，并通知给其他的客户端。这篇文章会深入探讨zookeeper的session管理机制。

zookeeper的心跳机制

为什么要有心跳机制

zookeeper底层支持两种网络库，一种是zookeeper基于NIO自己写的，一种是Netty。那么zookeeper能不能直接通过感知TCP连接是否断开来感知客户端连接是否断开呢？

答案是不能，原因有很多，个人觉得最重要的一点是，基于TCP连接来判断客户端是否存活是不靠谱的。

这里举两个异常case 1、客户端进程直接crash了，还没来得及发送FIN报文，这种情况下，zookeeper在TCP这一侧是没办法及时感知到TCP连接已经失效了。（也就是说，由于zookeeper没收到client的FIN包，虽然client已经挂了，TCP侧还认为客户端还活着。只能等弱鸡keepalive了）

2、客户端和服务端之间很久没有报文交互，TCP连接其实已经失效了。（失效原因很多，比如路由器出问题了，网络设备故障了等等）这时候，如果客户端发送报文给服务端，linux会进行重试，默认差不多要重试15分钟，才能感知到这个连接已经失效。

这里只是举个例子。实际来说，client和zookeeper的报文交互可能不会那么少。也就是第二种情况，客户端很久不给服务端发报文，要发了，才发现tcp连接已经有问题了，这种情况不太可能出现。（当然，选举这种场景是有可能出现这种情况的，后面会单独写文章分析这个case） 但是从上面两个例子，我们也不难得出结论，通过tcp连接的断开与否来感知客户端是否存活，似乎并不太靠谱。

client发送心跳

因为tcp的“不靠谱”，zookeeper为了能够实现可靠的连接管理（也为了保活），选择自己实现心跳机制。

正如上图所示，client隔一段时间就会发送一个心跳报文给服务端，告诉zookeeper自己还活着，别把我连接关了。

注意：这里乱入了一个create报文，因为正常报文也算是一种“心跳”。反正，zookeeper只要能收到报文，就能知道客户端还活着。

服务端收到报文后，会更新session信息

这里带大家看看代码。很简单，收到报文后，会调用 SessionTracker.touchSession() 来更新session信息

sessionId是zookeeper为每一个连接分配的唯一id

管理session的难点在哪

zookeeper 管理session的需求分析

这里我们思考下zookeeper对于session管理的需求是什么？

1、zookeeper需要有个地方存session

2、当客户端一段时间没发生心跳时，zookeeper要能感知到

第一个问题比较简单，java提供了各式各样的集合，无论是Map或者是List理论上都能存session。由于我们一般是通过sessionId来找到关联的Session的，因此使用Map更合适点。Zookeeper也是这么做的 protected final ConcurrentHashMap sessionsById 这里sessionsById 就是zookeeper管理Session使用的容器

第二个问题看起来也很简单，客户端不是会发送心跳么？我给每一个session记录一个上一个报文到达时间，一旦收到新的报文，我就更新这个时间。然后我再不断地扫描每个session是否很久没收到报文不就行了？

如何检测Session失效？

按照上面的说法，收到一个报文，我就更新Session的“上次报文”字段。假设session失效时间是4秒，我就每隔4秒扫描一次session的集合，找出那个超过4秒没有新的报文的session不就行了？像上图一样，假设sessionTimeout是4秒，现在已经11点50分05秒了，理论上每个session上一个报文时间应该大于11点50分01秒。很明显，session1失效了。

定时任务的选型

既然要定时扫描，我们就需要跑一个定时任务。jdk本身也提供了很多的定时任务方案。不知道定时任务的同学可以参考这篇文章Java中定时任务的6种实现方式，你知道几种？- 掘金 既然如此，我们完全可以使用jdk自带的定时任务，定时去扫描这个集合啊，这样不就能很轻易的找到失效的session了么？

这里有两个问题：1、每一个客户端和服务端的连接，sessionTimeout都是可配的。我们例子是4秒没收到报文，就认为连接失效。实际超时时间可能有1秒、2秒、3秒.... 所以如果用定时任务来实现，我们可能需要启动不止一个定时任务。2、jdk提供的定时任务不够灵活，什么意思呢。比如我设置的sessionTimeout是4秒，现在是11点。然后我在11点00分02秒就收到了一个心跳，那么下次检测时间应该变成11点00分06秒。而jdk的定时任务限定了，只能每隔4秒检测一次。比如：11点、11点00分04秒、11点00分08秒、这样检测下去。而如果用jdk的定时任务，我们只能简单的隔一段时间，检测一次。这里可以仔细体会下两者的差异。

想一想这些问题，是不是发现想实现一个高效的session管理机制是不是没那么简单。接下来我们看看，zookeeper是如何巧妙地实现session的管理。

zookeeper session管理机制

接下来就要介绍zookeeper的session管理机制了。

1、通过expiryMap存储过期时间与session集合的对应关系

首先，zookeeper内部有一个expiryMap

非常简单，Key是过期时间，value是一个Set，里面放了一个个的Session。

举个例子：S1在11点50分02秒的key下面，表示如果11点50分02秒前没收到新的报文，就认为S1过期了。

2、当收到某条连接的报文时，更新expiryMap

拿上图的S1为例，它的sessionTimeout是4秒，在11点50分01秒收到报文。那么理论上下个session检测时间会是 11点50分05秒。

这里要说下expiryMap的第一个特征，它的key并不是随意一个时间。它会间隔一个固定的时间叫做 expirationInterval，数值上它等于zookeeper的配置tickTime（默认配的2秒）

所以说，这里计算出11点50分05秒后，它会round下，round到11点50分06秒

如图所示：

S1在截止时间前更新了session，我们就要把它从旧的桶里移除，挪到新的桶里。

3、循环检测ExpiryMap

有个SessionTracker 线程会循环检测这个expiryMap，找到最近的那个key对应的session集合，把他们全部都过期掉。

就像上面的例子，一旦时间到了11点50分02秒，就把对应的session全部过期掉。

小结下

1、如果收到报文，会把session放到下一个过期桶里。2、SessionTracker会按次序，不断地取出过期的桶，把桶里的session全部过期掉（过期会删除临时节点，当然还有其他一系列操作） 3、zookeeper底层使用了非常简单的Map就实现了非常高效的Session管理机制。

session管理机制源码分析

接下来我们来看看源码

1、server端的心跳续约

 

//org.apache.zookeeper.server.ExpiryQueue#update
public Long update(E elem, int timeout) {
  //1、除了上面我们介绍的ExpiryMap，zookeeper内部还有一个elemMap，用于存放 Session -> 过期时间
  Long prevExpiryTime = elemMap.get(elem);
  long now = Time.currentElapsedTime();
  //2、收到心跳后，我们会计算session应该更新到哪个桶里
  Long newExpiryTime = roundToNextInterval(now + timeout);


  //桶不变，就不用更新expiryMap了
  if (newExpiryTime.equals(prevExpiryTime)) {
    // No change, so nothing to update
    return null;
  }


  // First add the elem to the new expiry time bucket in expiryMap.
  //3. 找到新的桶，插入进去
  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);
  //4. 从旧的桶里移除
  if (prevExpiryTime != null && !newExpiryTime.equals(prevExpiryTime)) {
    Set prevSet = expiryMap.get(prevExpiryTime);
    if (prevSet != null) {
      prevSet.remove(elem);
    }
  }
  return newExpiryTime;
}

 

其实就是这幅图 1、放入一个：Session -> 过期时间 的Map中 2、收到心跳后，我们会计算session应该更新到哪个桶里 3、找到新的桶，插入进去 4、把session从旧的桶里移除

2、SessionTracker不断地轮训，找到过期的Session集合，然后都过期掉

 

//org.apache.zookeeper.server.SessionTrackerImpl#run
@Override
public void run() {
  try {
    while (running) {
      //1. 这个其实就是不断地轮训下一个要检测的key
      // 比如按我们的例子，应该是11点50分02秒检测一次、11点50分04秒检测一次、11点50分06秒检测一次...
      // 这里的waitTime就是找到下次检测需要等待多久，比如现在是11点50分01秒了，这个waitTime就是1秒
      // 如果是11点50分02秒了，waitTime就是0，我们要开始把过期的session都失效掉了
      long waitTime = sessionExpiryQueue.getWaitTime();
      if (waitTime > 0) {
        Thread.sleep(waitTime);
        continue;
      }


      //2. 取出过期的Session集合，全部都expire掉，如果session都及时发送了心跳了，这里就会拿到一个空的集合
      for (SessionImpl s : sessionExpiryQueue.poll()) {
        ServerMetrics.getMetrics().STALE_SESSIONS_EXPIRED.add(1);
        setSessionClosing(s.sessionId);
        expirer.expire(s);
      }
    }
  } catch (InterruptedException e) {
    handleException(this.getName(), e);
  }
  LOG.info("SessionTrackerImpl exited loop!");
}

 

其实就是这个图

　　审核编辑：汤梓红