使用过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#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<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;
}
[图片上传失败...(image-6f8444-1704619490775)]
其实就是这幅图
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!");
}
其实就是这个图