深入详解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<Long, SessionImpl> 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#updatepublic 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<E> set = expiryMap.get(newExpiryTime); if (set == null) { // Construct a ConcurrentHashSet using a ConcurrentHashMap set = Collections.newSetFromMap(new ConcurrentHashMap<E, Boolean>()); // Put the new set in the map, but only if another thread // hasn't beaten us to it Set<E> 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<E> 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@Overridepublic 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!");}

其实就是这个图

链接:https://juejin.cn/post/7320898289187323958

(版权归原作者所有,侵删)

标签

发表评论

苏ICP备2023047577号-1