[ZooKeeper]分布式Session创建

7007

　　前面几篇zookeeper的文章简单分析了执行流程，接下来打算从横向来分析一下zk的一些特性，先从session开始。
这一篇http://iwinit.iteye.com/blog/1754611分析了单机情况下session建立，在集群环境下建立session不太一样，是一个proposal的过程，先假设集群由leader，followerA，followerB组成，我们的client去连followerA。follower和leader初始化之后，初始化的sessionTracker不一样，leader中是SessionTrackerImpl，follower中是LearnerSessionTracker，主要区别和类同点：
1.follower中不会启动超时检查线程，只是简单得记录了session信息，主要数据结构是

//session更新信息
HashMap<Long, Integer> touchTable = new HashMap<Long, Integer>();
//session超时信息
private ConcurrentHashMap<Long, Integer> sessionsWithTimeouts;
　　而leader会启动超时线程，而且数据结构也多一些

//session实体
HashMap<Long, SessionImpl> sessionsById = new HashMap<Long, SessionImpl>();
//同一个超时时间点的session，给超时线程用
HashMap<Long, SessionSet> sessionSets = new HashMap<Long, SessionSet>();
//session超时信息
ConcurrentHashMap<Long, Integer> sessionsWithTimeout;
　　2.addSession时，leader会创建session实体，follower只是简单的记录了一下session信息
　　3.sessionId初始化算法一样

//1字节server_id+当前时间的后5个字节+2字节0，保证全局唯一
public static long initializeNextSession(long id) {
long nextSid = 0;
nextSid = (System.currentTimeMillis() << 24) >> 8;
nextSid =  nextSid | (id <<56);
return nextSid;
}
　　连下来client开始创建session
　　1.客户端发送ConnectionRequest给followerA
2.followerA处理
　　session超时时间协商

//处理session时间，minSessionTimeout为ticktime2倍，maxSessionTimeout为ticktime20倍
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();
//客户端发送的sessionid，重试时不为0
long sessionId = connReq.getSessionId();
//客户端重试，则reopen，后文分析
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);
}
//创建session
else {
LOG.info("Client attempting to establish new session at "
+ cnxn.getRemoteSocketAddress());
createSession(cnxn, passwd, sessionTimeout);
}
　　sessionid和密码初始化

long createSession(ServerCnxn cnxn, byte passwd[], int timeout) {
//sessionid递增
long sessionId = sessionTracker.createSession(timeout);
//随机密码
Random r = new Random(sessionId ^ superSecret);
r.nextBytes(passwd);
//4个字节的超时时间
ByteBuffer to = ByteBuffer.allocate(4);
to.putInt(timeout);
cnxn.setSessionId(sessionId);
//异步提交执行链
submitRequest(cnxn, sessionId, OpCode.createSession, 0, to, null);
return sessionId;
}
　　提交请求

   private void submitRequest(ServerCnxn cnxn, long sessionId, int type,
int xid, ByteBuffer bb, List<Id> authInfo) {
//初始化时，xid为0，bb为4个字节的session超时时间
Request si = new Request(cnxn, sessionId, xid, type, bb, authInfo);
submitRequest(si);
}
　　根据之前http://iwinit.iteye.com/blog/1777109分析的Processor链图，FollowerRequestProcessor执行。和create请求一样，createSession是事务请求需要投票，FollowerA发送投票packet给leader。
　　3.leader处理，learnerHandler中收到request请求，提交leader的processor链
　　PrepRequestProcessor处理

//头信息
request.hdr = new TxnHeader(request.sessionId, request.cxid, zxid,
zks.getTime(), type);
....
case OpCode.createSession:
request.request.rewind();
int to = request.request.getInt();
//txn信息，就是一个session超时
request.txn = new CreateSessionTxn(to);
request.request.rewind();
//leader中创建session
zks.sessionTracker.addSession(request.sessionId, to);
//owner属于followerA
zks.setOwner(request.sessionId, request.getOwner());
break;
　　之后ProposalRequestProcessor发起投票，并写入log
　　4.followerA和followerB处理投票

public void logRequest(TxnHeader hdr, Record txn) {
Request request = new Request(null, hdr.getClientId(), hdr.getCxid(),
hdr.getType(), null, null);
request.hdr = hdr;
request.txn = txn;
request.zxid = hdr.getZxid();
//添加到pending队列
if ((request.zxid & 0xffffffffL) != 0) {
pendingTxns.add(request);
}
/写入log，并发送ack包给leader
syncProcessor.processRequest(request);
}
　　5.leader收到投票，发起commit，然后自己commit
　　6.leader的FinalRequestProcessor处理，添加session，不需要返回client数据

public ProcessTxnResult processTxn(TxnHeader hdr, Record txn) {
ProcessTxnResult rc;
int opCode = hdr.getType();
long sessionId = hdr.getClientId();
//createSession不需要修改db状态，啥都不做
rc = getZKDatabase().processTxn(hdr, txn);
//又添加了一次session
if (opCode == OpCode.createSession) {
if (txn instanceof CreateSessionTxn) {
CreateSessionTxn cst = (CreateSessionTxn) txn;
sessionTracker.addSession(sessionId, cst
.getTimeOut());
} else {
LOG.warn("*****>>>>> Got "
+ txn.getClass() + " "
+ txn.toString());
}
} else if (opCode == OpCode.closeSession) {
sessionTracker.removeSession(sessionId);
}
return rc;
}
　　7.followerA和followerB处理commit，区别在于CommitRequestProcessor中，followerA中的Request会带上connection信息而followerB中的reqesut没有connection信息。所以在FinalRequestProcessor中，followerB创立完session就返回了，而followerA还需要写回client响应
　　最后看下leader的sessionTracker超时机制，构造SessionTrackerImpl

public SessionTrackerImpl(SessionExpirer expirer,
ConcurrentHashMap<Long, Integer> sessionsWithTimeout, int tickTime,
long sid)
{
super("SessionTracker");
this.expirer = expirer;
this.expirationInterval = tickTime;
this.sessionsWithTimeout = sessionsWithTimeout;
//下一个检查点，向上取整为expirationInterval倍数，expirationInterval就是配置的ticktime
nextExpirationTime = roundToInterval(System.currentTimeMillis());
this.nextSessionId = initializeNextSession(sid);
for (Entry<Long, Integer> e : sessionsWithTimeout.entrySet()) {
addSession(e.getKey(), e.getValue());
}
}
　　主线程循环

    synchronized public void run() {
try {
while (running) {
//下一个超时点没到，就等待
currentTime = System.currentTimeMillis();
if (nextExpirationTime > currentTime) {
this.wait(nextExpirationTime - currentTime);
continue;
}
//同一个时间点超时的session
SessionSet set;
set = sessionSets.remove(nextExpirationTime);
//超时session处理
if (set != null) {
for (SessionImpl s : set.sessions) {
setSessionClosing(s.sessionId);
expirer.expire(s);
}
}
//下一个check point
nextExpirationTime += expirationInterval;
}
} catch (InterruptedException e) {
LOG.error("Unexpected interruption", e);
}
LOG.info("SessionTrackerImpl exited loop!");
}
　　session更新时

    synchronized public boolean touchSession(long sessionId, int timeout) {
......
//session对象
SessionImpl s = sessionsById.get(sessionId);
// Return false, if the session doesn't exists or marked as closing
if (s == null || s.isClosing()) {
return false;
}
//这个session的下一个超时点，向上取整为ticktime倍数
long expireTime = roundToInterval(System.currentTimeMillis() + timeout);
//时间点比老的时间还小，不更新
if (s.tickTime >= expireTime) {
// Nothing needs to be done
return true;
}
//先从老的超时set中remove掉，再添加到新的set中，超时线程会定时check
SessionSet set = sessionSets.get(s.tickTime);
if (set != null) {
set.sessions.remove(s);
}
//下个超时点
s.tickTime = expireTime;
//添加到新的超时set中
set = sessionSets.get(s.tickTime);
if (set == null) {
set = new SessionSet();
sessionSets.put(expireTime, set);
}
set.sessions.add(s);
return true;
}
　　简单小节
1.session创建需要投票处理
2.结果是每台server上的内存中都会建立相同的session记录
3.sessionid通过serverid+时间保证唯一
4.session超时检查由leader负责，以ticktime定时检查
5.session更新时，会修改自己这个session所属的超时set，超时时间是ticktime倍数