Zookeeper Session机制

1、zookeeper session是什么

    客户端与服务端的任何交换操作都与会话相关，包括：临时节点的生命周期，客户端的请求顺序执行以及Watcher通知机制。

 2、客户端操作

2.1、Zookeeper客户端一次创建会话的过程。

 

2.2、构造会话请求

long sessId = (seenRwServerBefore) ? sessionId : 0;
ConnectRequest conReq = new ConnectRequest(0, lastZxid,sessionTimeout, sessId, sessionPasswd);
outgoingQueue.addFirst(new Packet(null, null, conReq,null, null, readOnly));

 

2.3、处理Response

ConnectResponse conRsp = new ConnectResponse();
conRsp.deserialize(bbia, "connect");
this.sessionId = conRsp.getSessionId();
sendThread.onConnected(conRsp.getTimeOut(),this.sessionId,conRsp.getPasswd(), isRO);

2.3.1、更新客户端Timeout等信息

void onConnected(int _negotiatedSessionTimeout, long _sessionId,
                byte[] _sessionPasswd, boolean isRO) throws IOException {
            negotiatedSessionTimeout = _negotiatedSessionTimeout;
            if (negotiatedSessionTimeout <= 0) {
                state = States.CLOSED;

                eventThread.queueEvent(new WatchedEvent(
                        Watcher.Event.EventType.None,
                        Watcher.Event.KeeperState.Expired, null));
                eventThread.queueEventOfDeath();
                throw new SessionExpiredException(
                        "Unable to reconnect to ZooKeeper service, session 0x"
                                + Long.toHexString(sessionId) + " has expired");
            }
            if (!readOnly && isRO) {
                LOG.error("Read/write client got connected to read-only server");
            }
            readTimeout = negotiatedSessionTimeout * 2 / 3;
            connectTimeout = negotiatedSessionTimeout / hostProvider.size();
            hostProvider.onConnected();
            sessionId = _sessionId;
            sessionPasswd = _sessionPasswd;
            state = (isRO) ?
                    States.CONNECTEDREADONLY : States.CONNECTED;
            seenRwServerBefore |= !isRO;
            LOG.info("Session establishment complete on server "
                    + clientCnxnSocket.getRemoteSocketAddress()
                    + ", sessionid = 0x" + Long.toHexString(sessionId)
                    + ", negotiated timeout = " + negotiatedSessionTimeout
                    + (isRO ? " (READ-ONLY mode)" : ""));
            KeeperState eventState = (isRO) ?
                    KeeperState.ConnectedReadOnly : KeeperState.SyncConnected;
            eventThread.queueEvent(new WatchedEvent(
                    Watcher.Event.EventType.None,
                    eventState, null));
        }

 

2.4、发送Ping请求

    客户端当sessionTimeout/3(readTimeout=sessionTimeout * 2 /3 )之内都没有发送通信请求后，会主动发起一个ping请求，以维持会话状态.

if (state.isConnected()) {
    int timeToNextPing = readTimeout / 2
                          - clientCnxnSocket.getIdleSend();
   if (timeToNextPing <= 0) {
             sendPing();
            clientCnxnSocket.updateLastSend();
    } else {
             if (timeToNextPing < to) {
                to = timeToNextPing;
              }
     }
}

 

 3、服务端操作 

3.1、服务端会话处理流程示意图

    

 

3.2、协商sessionTimeout

    2Ticktime <= sessionTimeout <= 20Ticktime（tickTime默认等于2s）

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);

 

3.3、判断是否需要创建新会话

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);
  }

 

3.4、创建sessionId

     每次客户端创建新会话时，zookeeper都会为其分配一个全局唯一的sessionID。

public static long initializeNextSession(long id) {
        long nextSid = 0;
        nextSid = (System.currentTimeMillis() << 24) >> 8;
        nextSid =  nextSid | (id <<56);
        return nextSid;
}
synchronized public long createSession(int sessionTimeout) {
     addSession(nextSessionId, sessionTimeout);
     return nextSessionId++;
}

 

 

 3.5、注册会话

 

SessionTracker中维护两个数据结构sessionsWithTimeout和sessionsById，前者根据sessionID保存了所有会话超时时间，后者则根据sessionID保存了会话实体。

 

  HashMap<Long, SessionImpl> sessionsById = new HashMap<Long, SessionImpl>();
  ConcurrentHashMap<Long, Integer> sessionsWithTimeout; 

   synchronized public void addSession(long id, int sessionTimeout) {
        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);
            }
        } else {
            if (LOG.isTraceEnabled()) {
                ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
                        "SessionTrackerImpl --- Existing session 0x"
                        + Long.toHexString(id) + " " + sessionTimeout);
            }
        }
        touchSession(id, sessionTimeout);
    }

 

 3.6、激活会话

 

  3.6.1、分桶策略

   

将类似的会话放在同一区块中进行管理，以便于不同区块的隔离，相同区块的统一处理。

   分配的原则：会话的下一次超时的时间点（ExpirationTime, ExpirationInterval=tickTime 单位：毫秒）。

   ExpirationTime = CurrentTime + SessionTimeout

   ExpirationTime =  (ExpirationTime / ExpirationInterval +1) x ExpirationInterval

   以ExpirationTime为key存入Map中，实现了分桶。

 

 3.6.2、激活

 

   客户端会向服务端发送ping请求来保持会话的有效，服务端收到心跳会不断的更新对应的客户端会话。

  

HashMap<Long, SessionSet> sessionSets = new HashMap<Long, SessionSet>();

private long roundToInterval(long time) {
        // We give a one interval grace period
        return (time / expirationInterval + 1) * expirationInterval;
}
 synchronized public boolean touchSession(long sessionId, int timeout) {
        if (LOG.isTraceEnabled()) {
            ZooTrace.logTraceMessage(LOG,
                                     ZooTrace.CLIENT_PING_TRACE_MASK,
                                     "SessionTrackerImpl --- Touch session: 0x"
                    + Long.toHexString(sessionId) + " with timeout " + timeout);
        }
        SessionImpl s = sessionsById.get(sessionId);
        // Return false, if the session doesn't exists or marked as closing
        if (s == null || s.isClosing()) {
            return false;
        }
        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);
        return true;
    }

 

3.7、会话清理

 

3.7.1、会话超时检测

       分桶策略的数据迁移是被动触发，没有做迁移的sessionId会一直保存在之前的桶中，SessionTracker线程会检测到过期(没有迁移的)的会话。

 

3.7.2、清理过程

1、标记会话为“已关闭”

2、发起“会话关闭”请求

3、收集要清理的临时节点

4、添加“节点删除”事务

5、删除临时节点

6、移除会话

7、关闭NioServerCnxn

 

    synchronized public void run() {
        try {
            while (running) {
                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) {
                        setSessionClosing(s.sessionId);
                        expirer.expire(s);
                    }
                }
                nextExpirationTime += expirationInterval;
            }
        } catch (InterruptedException e) {
            LOG.error("Unexpected interruption", e);
        }
        LOG.info("SessionTrackerImpl exited loop!");
    }

    synchronized public void setSessionClosing(long sessionId) {
        if (LOG.isTraceEnabled()) {
            LOG.info("Session closing: 0x" + Long.toHexString(sessionId));
        }
        SessionImpl s = sessionsById.get(sessionId);
        if (s == null) {
            return;
        }
        s.isClosing = true;
    }

    public void expire(Session session) {
        long sessionId = session.getSessionId();
        LOG.info("Expiring session 0x" + Long.toHexString(sessionId)
                + ", timeout of " + session.getTimeout() + "ms exceeded");
        close(sessionId);
    }

    private void close(long sessionId) {
        submitRequest(null, sessionId, OpCode.closeSession, 0, null, null);
    }

    

   

PrepRequestProcessor.pRequest2Txn(int type, long zxid, Request request, Record record, boolean deserialize)
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;

 

 

 

FinalRequestProcessor.processRequest(Request request) 
       if (request.hdr != null && request.hdr.getType() == OpCode.closeSession) {
            ServerCnxnFactory scxn = zks.getServerCnxnFactory();
            // this might be possible since
            // we might just be playing diffs from the leader
            if (scxn != null && request.cnxn == null) {
                // calling this if we have the cnxn results in the client's
                // close session response being lost - we've already closed
                // the session/socket here before we can send the closeSession
                // in the switch block below
                scxn.closeSession(request.sessionId);
                return;
            }
        }