[ZooKeeper]分布式Session创建

iwinit

浏览: 454798 次
性别:
来自: 杭州

最近访客更多访客>>

lingxiajiudu

plisking

rochine123

尼欧-张

博主相关

博客

微博

相册

留言

关于我

文章分类

全部博客 (45)

社区版块

存档分类

zookeeper session

前面几篇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倍数

分享到：