说到zookeeper session管理 ,免不了要问
- 什么是session?
- session id/session是如何产生的?
- session 信息如何存储?
本文以session tracker线程【详见SessionTrackerImpl】的运行机制作为主线,并尝试解答一些相关问题
1)session基础
在介绍session tracker线程之前先回答几个问题
1.1) 什么是session?
zookeeper中session意味着一个物理连接,客户端connect成功之后,会发送一个connect型请求,此时就会有session 产生(下面会具体讲)
1.2)sessionid是如何产生的?
在SessionTrackerImpl实例化的时候就会调用下面的函数【详见SessionTrackerImpl.initializeNextSession】
1
2
3
4
5
6
|
public static long initializeNextSession( long id) {
long nextSid = 0 ;
nextSid = (System.currentTimeMillis() << 24 ) >> 8 ;
nextSid = nextSid | (id << 56 );
return nextSid;
}
|
产生的值会存入nextSessionId属性,以后一旦有新的连接(session)产生,就会nextSessionId++
1.3)session是如何产生的?
接到一个连接类型的请求【详见ZooKeeperServer.processConnectRequest】
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
int sessionTimeout = connReq.getTimeOut();
byte passwd[] = connReq.getPasswd();
int minSessionTimeout = getMinSessionTimeout();
if (sessionTimeout < minSessionTimeout) {
sessionTimeout = minSessionTimeout;
}
int maxSessionTimeout = getMaxSessionTimeout();
if (sessionTimeout > maxSessionTimeout) {
sessionTimeout = maxSessionTimeout;
}
cnxn.setSessionTimeout(sessionTimeout);
// We don‘t want to receive any packets until we are sure that the
// session is setup
cnxn.disableRecv();
long sessionId = connReq.getSessionId();
if (sessionId != 0 ) {
long clientSessionId = connReq.getSessionId();
LOG.info( "Client attempting to renew session 0x"
+ Long.toHexString(clientSessionId)
+ " at " + cnxn.getRemoteSocketAddress());
serverCnxnFactory.closeSession(sessionId);
cnxn.setSessionId(sessionId);
reopenSession(cnxn, sessionId, passwd, sessionTimeout);
} else {
LOG.info( "Client attempting to establish new session at "
+ cnxn.getRemoteSocketAddress());
createSession(cnxn, passwd, sessionTimeout);
}
|
1.3.1)确定session的timeout和id
【详见SessionTrackerImpl.createSession】
1
2
3
4
|
synchronized public long createSession( int sessionTimeout) {
addSession(nextSessionId, sessionTimeout);
return nextSessionId++;
}
|
可见产生session需要两个元素,一个是sessionid,一个是timeout
- timeout由客户端确定,但必须在服务器规定的最大的timeout(ticktime*20)和最小的timeout(ticktime*2)之间
- 如果客户端没有指定sessionid,那么就会产生一个新的session【详见ZooKeeperServer.createSession】,否则会reopen【详见ZooKeeperServer.reopenSession】
- sessionid的产生上面解释过了
1.3.2)实例化session及相关关系存放
【详见SessionTrackerImpl.addSession】
1
2
3
4
5
6
7
8
9
10
|
sessionsWithTimeout.put(id, sessionTimeout);
if (sessionsById.get(id) == null ) {
SessionImpl s = new SessionImpl(id, sessionTimeout, 0 );
sessionsById.put(id, s);
if (LOG.isTraceEnabled()) {
ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
"SessionTrackerImpl --- Adding session 0x"
+ Long.toHexString(id) + " " + sessionTimeout);
}
}
|
- 一个重要的数据结构sessionsWithTimeout存放sessionid和timeout的映射
- 另一个重要的数据结构sessionsById存放sessionid和SessionImpl实例的映射
1.3.3)确定session实例的tickTime及sessionSets关系维护
【详见SessionTrackerImpl.touchSession】
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
long expireTime = roundToInterval(System.currentTimeMillis() + timeout);
if (s.tickTime >= expireTime) {
// Nothing needs to be done
return true ;
}
SessionSet set = sessionSets.get(s.tickTime);
if (set != null ) {
set.sessions.remove(s);
}
s.tickTime = expireTime;
set = sessionSets.get(s.tickTime);
if (set == null ) {
set = new SessionSet();
sessionSets.put(expireTime, set);
}
set.sessions.add(s);
|
- 根据当前时间和timeout计算本session 的expireTime即tickTime
- 一个重要的数据结构sessionSets 存放过期时间和一组session实例(相同过期时间)的映射的建立及维护
- session实例的tickTime的确定
2)session tracker线程的机制
在zookeeper服务体系中,专门有一个线程(session tracker)维护session【详见SessionTrackerImpl.run】,重要代码如下
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
currentTime = System.currentTimeMillis();
if (nextExpirationTime > currentTime) {
this .wait(nextExpirationTime - currentTime);
continue ;
}
SessionSet set;
set = sessionSets.remove(nextExpirationTime);
if (set != null ) {
for (SessionImpl s : set.sessions) {
sessionsById.remove(s.sessionId);
expirer.expire(s);
}
}
nextExpirationTime += expirationInterval;
|
可见SessionTrackerImpl这个线程会一直轮询的清除过期session
- 每次轮询都会比较currentTime和nextExpirationTime,如果还未到nextExpirationTime,就等,否则往下走
- 将sessionSets中的以nextExpirationTime为key的那组session移出
- 遍历session,从sessionsById移除session,并调用相关的过期处理(下面会讲)
- 调整下载比较的时间,即nextExpirationTime += expirationInterval;
3) session维护相关问题
3.1)清除session如何实现?
【详见ZooKeeperServer.close】
1
2
3
|
private void close( long sessionId) {
submitRequest( null , sessionId, OpCode.closeSession, 0 , null , null );
}
|
3.1.1)构造一个Request实例
3.1.2)调用PrepRequestProcessor.processRequest放入submittedRequests队列
3.1.3)PrepRequestProcessor线程的处理
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
request.hdr = new TxnHeader(request.sessionId, request.cxid, zxid,
zks.getTime(), type);
switch (type) {
//省略N行代码......
case OpCode.closeSession:
// We don‘t want to do this check since the session expiration thread
// queues up this operation without being the session owner.
// this request is the last of the session so it should be ok
//zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
HashSet<String> es = zks.getZKDatabase()
.getEphemerals(request.sessionId);
synchronized (zks.outstandingChanges) {
for (ChangeRecord c : zks.outstandingChanges) {
if (c.stat == null ) {
// Doing a delete
es.remove(c.path);
} else if (c.stat.getEphemeralOwner() == request.sessionId) {
es.add(c.path);
}
}
for (String path2Delete : es) {
addChangeRecord( new ChangeRecord(request.hdr.getZxid(),
path2Delete, null , 0 , null ));
}
zks.sessionTracker.setSessionClosing(request.sessionId);
}
LOG.info( "Processed session termination for sessionid: 0x"
+ Long.toHexString(request.sessionId));
break ;
|
- 设置request.hdr,这个很重要,
- 在FinalRequestProcessor.processRequest会有相应的处理
1
2
3
4
5
6
|
if (request.hdr != null ) {
TxnHeader hdr = request.hdr;
Record txn = request.txn;
rc = zks.processTxn(hdr, txn);
}
|
?
- 一旦某个session关闭,与session相关的EPHEMERAL类型的节点都得清除
- 并且通过调用sessionTracker.setSessionClosing将session设置为关闭,使得后续此session上的请求无效
3.1.4)SessionTrackerImpl相关数据结构的清理
【详见SessionTrackerImpl.removeSession】
1
2
3
4
5
6
7
8
9
10
11
12
|
synchronized public void removeSession( long sessionId) {
SessionImpl s = sessionsById.remove(sessionId);
sessionsWithTimeout.remove(sessionId);
if (LOG.isTraceEnabled()) {
ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
"SessionTrackerImpl --- Removing session 0x"
+ Long.toHexString(sessionId));
}
if (s != null ) {
sessionSets.get(s.tickTime).sessions.remove(s);
}
}
|
分别对sessionsById、sessionsWithTimeout、sessionSets进行处理
3.2)session owner咋回事?
如果不是在集群环境,即没有LearnerHandler线程,session 的owner就是一个常量实例ServerCnxn.me
3.3)sessionsWithTimeout这个数据结构的用途?
sessionsWithTimeout存放的是sessionid和timeout,此数据结构会和ZKDatabase中相通,会被持久化
如果某个session timeout为60s,如果空闲了30s,意味着还能空闲30s,此时服务重启,那么此session的timeout又变为60s
3.4)touch session是干吗的?
每次一旦该session有请求,就会touch,意味着session的过期时间变为(基本等于当前时间+timeout)
具体算法为
1
2
3
4
|
private long roundToInterval( long time) {
// We give a one interval grace period
return (time / expirationInterval + 1 ) * expirationInterval;
}
|
time为System.currentTimeMillis() + timeout
expirationInterval默认为ticktime
3.5)check session是干吗的
基本上所有的事务型操作,都会调用用来验证当前请求的session是否关闭,owner是否正确
4)小结
- SessionTrackerImpl作为一个单独的线程专门处理过期session
- SessionTrackerImpl有3个重要的数据结构sessionsById、sessionSets、sessionsWithTimeout,其中sessionsWithTimeout会被持久化
- SessionTrackerImpl提供了几个常用的API
- createSession
- addSession
- touchSession
- removeSession
- checkSession
- setOwner
- dumpSessions
时间: 2024-10-06 12:44:32