Zookeeper會話管理原始碼深入探討-有解無憂

首發CSDN：徐同學呀，原創不易，轉載請注明源鏈接，我是徐同學，用心輸出高質量文章，希望對你有所幫助，

文章目錄

- SessionTracker類關系結構
- - LearnerSessionTracker
  - - touchTable
  - LeaderSessionTracker
- 創建sessionId
- 過期時間分桶管理會話
- - 理論
  - 原始碼實作
- 會話過期檢查
- 會話清理
- 會話激活

SessionTracker類關系結構

Zookeeper服務器會話管理是由SessionTracker完成的，

SessionTracker類關系結構

SessionTracker由ZooKeeperServer所持有，具體持有情況如下：

LearnerZooKeeperServer 持有 LearnerSessionTracker，
LeaderZooKeeperServer 持有 LeaderSessionTracker，
LearnerSessionTracker 和 LeaderSessionTracker 都可以持有 LocalSessionTracker，

LearnerSessionTracker

LearnerSessionTracker只負責session的創建，并不對session做過期檢查、清理等作業，

touchTable

LearnerSessionTracker用AtomicReference<Map<Long, Integer>> touchTable存盤sessionId和timeout的關系，Map<Long, Integer>是ConcurrentHashMap，key是sessionId，value是timeout，

touchTable相關增、刪、改、查方法分別為：

LearnerSessionTracker#commitSession，創建session的事務請求在兩階段的commit階段時呼叫，
LearnerSessionTracker#removeSession，關閉session時呼叫，
LearnerSessionTracker#touchSession，客戶端向服務端發送請求的流程中呼叫，用于重置或者添加session，
LearnerSessionTracker#snapshot，獲取整個Map并將touchTable置空，因為Learner不管理session，所以在Learner接收到Leader的心跳時，會將持有的所有session（sessionId+timeout）取出來構建成PING回應包發給Leader，通知Leader更新對應session的過期時間，為了保證操作touchTable時執行緒安全，故而將其設定成AtomicReference修飾的，

public Map<Long, Integer> snapshot() {
    return touchTable.getAndSet(new ConcurrentHashMap<Long, Integer>());
}

LeaderSessionTracker

LeaderSessionTracker中對session的管理是委托給globalSessionTracker的，globalSessionTracker是SessionTrackerImpl的實體物件，其是一個執行緒，

每一個會話在SessionTrackerImpl內部都保留了三份：

sessionsById：HashMap＜Long，SessionImpl＞型別的資料結構，用于根據sessionID來管理Session物體SessionImpl，
sessionsWithTimeout：ConcurrentHashMap＜Long，Integer＞型別的資料結構，用于根據 sessionID 來管理會話的超時時間，該資料結構和ZooKeeper記憶體資料庫相連通，會被定期持久化到快照檔案中去，
sessionExpiryQueue：ExpiryQueue<SessionImpl>定制化實作的按過期時間分桶管理會話的資料結構，

創建sessionId

創建session實則是創建sessionId，

// org.apache.zookeeper.server.quorum.LearnerSessionTracker#createSession
public long createSession(int sessionTimeout) {
    if (localSessionsEnabled) {
        return localSessionTracker.createSession(sessionTimeout);
    }
    return nextSessionId.getAndIncrement();
}

sessionId會在LearnerSessionTracker創建時呼叫initializeNextSessionId設定一個初始值，后面就是在這個初始值的基礎上自增來分配sessionID，

// org.apache.zookeeper.server.SessionTrackerImpl#initializeNextSessionId
public static long initializeNextSessionId(long id) {
    long nextSid;
    nextSid = (Time.currentElapsedTime() << 24) >>> 8;
    nextSid = nextSid | (id << 56);
    if (nextSid == EphemeralType.CONTAINER_EPHEMERAL_OWNER) {
        ++nextSid;  // this is an unlikely edge case, but check it just in case
    }
    return nextSid;
}

從原始碼可以總結出，生成sessionId初始值一共分為5個步驟：
假設當前時間毫秒值為1644397769145，機器sid為1，
（1）獲取當前時間毫秒值

0000 0000 0000 0000 0000 0001 0111 1110 1101 1101 1011 1110 1011 0001 1011 1001

（2）向左移動24位，右邊24個0就是左移出來的，
左移24位的目的是，將高位的1移出，剩下的最高位是0，這樣得到的數就是一個正數，

0111 1110 1101 1101 1011 1110 1011 0001 1011 1001 0000 0000 0000 0000 0000 0000

當前時間毫秒值左移24位不一定是個正數，2022-04-07 01:50:41:664 之后的第40位是1，左移24位后，高位就是1，顯然是個負數，所以才有了右移8位是無符號的，

2022-04-07 01:50:41:664
0000 0000 0000 0000 0000 0001 1000 0000 0000 0000 0000 0000 0000 0000 0000 0000

（3）無符號右移8位
>>>表示無符號右移，也叫邏輯右移，即若該數為正，則高位補0，而若該數為負數，則右移后高位同樣補0，

0000 0000 0111 1110 1101 1101 1011 1110 1011 0001 1011 1001 0000 0000 0000 0000

（4）機器標識 SID 左移 56位

0000 0001 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000

（5）（3）計算的數值和（4）計算的資料做 | 運算
| : 按位或，雙目運算，1 | 1 = 1，1 | 0 = 1，0 | 1 = 1，0 | 0 = 0

0000 0001 0111 1110 1101 1101 1011 1110 1011 0001 1011 1001 0000 0000 0000 0000

從sessionId創建可以看出，sessionId共64位，高8位是機器標識sid，中間40位是時間毫秒值，低16位是0，可用于后續自增，

1、先來看高8位是機器標識SID，實際生產中集群節點數最大可以是多少個？
原始碼中對機器標識SID限制不多，只有一個不能等于-1，其他數值經測驗SID可以為負數，0，正數：

SID取值為負數生成sessionId就沒有多大意義了，難以區分出高8位是SID；
SID也一般不設定為0，0和256重復了，左移56位都是0；
SID取值大于256也沒有意義，左移56位都是0，這樣sessionId高8位沒有識別度，可能會全域不唯一；
SID在[1, 127]范圍內，計算的sessionId是正數，一個zk集群有127個節點也完全足夠；
SID在[128,255]范圍內，第八位是1，左移56位左移56位就是一個負數，所以最后計算的sessionId也是一個負數，

所以一般SID正常取值為[1, 127]，

2、再來看中間40位的時間毫秒值，是否存在(時間毫秒數<<24) >>> 8 計算的值有相等的情況？

通過(時間毫秒數<<24) >>> 8方式計算結果范圍在[0, 1099511627776) （1L << 40=1099511627776），也就是兩個時間戳之間間隔1099511627776毫秒，計算結果就會有重復，1099511627776毫秒換算成年，差不多是34.8年，所以計算的40位的時間毫秒值，每34.8年重復一次，

3、最后看低16位自增是否夠用？
16位用于自增，取值范圍為[0, 65535]，也就是說1ms最多可以生成65536個sessionId，1秒6553.6萬，理論qps就是6553.6萬/s，這個qps大部分業務應該是夠的，

過期時間分桶管理會話

理論

所謂按過期時間分桶管理會話，就是最近一次具有相同過期時間點的會話放在一個桶里，這樣方便批量管理，最近一次過期時間點如何計算？

nextExpirationTime = currentTime + sessionTimeout

如果真以這種方式計算最近一次過期時間點的話，桶會分的很散，看看zookssper是如何分桶的？

ZooKeeper的Leader服務器在運行期間會啟動一個執行緒（globalSessionTracker）定時地進行會話過期檢查，其時間間隔是expirationInterval，單位是毫秒，默認是 tickTime，即默認情況下，每隔 2000 毫秒進行一次會話超時檢查，

為了方便對多個會話同時進行超時檢查，完整的nextExpirationTime的計算方式如下：

time = currentTime + sessionTimeout;
nextExpirationTime = (time / expirationInterval + 1) * expirationInterval;

如上方式計算的nextExpirationTime比正常計算的過期時間多出 (0,2000] ms，且總是expirationInterval的整數倍，
假設當前時間為1644377661000，sessionTimeout為20000：

time = 1644377661000 + 20000 = 1644377681000
nextExpirationTime = (1644377681000/2000 + 1) * 2000
nextExpirationTime = (822188840 + 1) * 2000
nextExpirationTime = 1644377682000

計算的nextExpirationTime比正常過期時間time多1000ms，

因此會話過期時長實際為(sessionTimeout，sessionTimeout + expirationInterval] ms，因此，將要在這個范圍內過期的會話會放在一個桶里管理，

用expirationInterval的倍數作為時間點來分布會話，因此，超時檢查執行緒只要在這些指定的時間點上進行檢查即可，

原始碼實作

分桶管理實則是一個Map資料結構，key為nextExpirationTime，value為Set就是裝會話實體的桶，

// E 泛型 可對應 SessionImpl
private final ConcurrentHashMap<Long, Set<E>> expiryMap = new ConcurrentHashMap<Long, Set<E>>();

為了方便通過會話檢索nextExpirationTime，即找到會話所在的桶編號，還維護了一個Map，key為會話實體，value為nextExpirationTime，

private final ConcurrentHashMap<E, Long> elemMap = new ConcurrentHashMap<E, Long>();

計算 nextExpirationTime：

private long roundToNextInterval(long time) {
    // nextExpirationTime值總是ExpirationInterval的整數倍數
    // time / expirationInterval 會向下取整，+1就是向上取整，寧可大也不能比原來小
    return (time / expirationInterval + 1) * expirationInterval;
}

呼叫示例：

Long newExpiryTime = roundToNextInterval(now + timeout)

會話過期檢查

LeaderSessionTracker會啟動一個超時檢查執行緒回圈定時檢查會話桶是否過期，過期的則進行清理操作，

// org.apache.zookeeper.server.SessionTrackerImpl#run
public void run() {
    try {
        while (running) {
            // 判斷 下一個檢查過期時間點是否到了
            // waitTime = expirationTime - now
            long waitTime = sessionExpiryQueue.getWaitTime();
            if (waitTime > 0) {
                // 沒到就休眠
                Thread.sleep(waitTime);
                continue;
            }
            // 依次清理過期 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!");
}

獲取距離下次過期時間間隔，waitTime > 0 則休眠waitTime時間：

// org.apache.zookeeper.server.ExpiryQueue#getWaitTime
public long getWaitTime() {
    long now = Time.currentElapsedTime();
    long expirationTime = nextExpirationTime.get();
    return now < expirationTime ? (expirationTime - now) : 0L;
}

從 ExpiryQueue 中取出過期的會話桶：

// org.apache.zookeeper.server.ExpiryQueue#poll
public Set<E> poll() {
    long now = Time.currentElapsedTime();
    // 1. 判斷 expirationTime 是否過期
    long expirationTime = nextExpirationTime.get();
    if (now < expirationTime) {
        return Collections.emptySet();
    }

    Set<E> set = null;
    long newExpirationTime = expirationTime + expirationInterval;
    // 2. cas 更新 nextExpirationTime
    if (nextExpirationTime.compareAndSet(expirationTime, newExpirationTime)) {
        // 3. 清理 過期桶
        set = expiryMap.remove(expirationTime);
    }
    if (set == null) {
        return Collections.emptySet();
    }
    return set;
}

會話清理

檢測到會話過期，需要對該會話做清理作業，
會話超時執行緒

1、標記會話狀態為“正在關閉”：
由于整個會話清理程序需要一段的時間，因此為了保證在此期間不再處理來自該客戶端的新請求，SessionTracker會首先將該會話的isClosing屬性標記為true，這樣，即使在會話清理期間接收到該客戶端的新請求，也無法繼續處理了，

2、發起發起“會話關閉”請求：
為了使對該會話的關閉操作在整個服務端集群中都生效，ZooKeeper 使用了提交“會話關閉”請求的方式，并立即交付給PrepRequestProcessor處理器進行處理，

過期的會話桶中的會話依次構建并發起“會話關閉”請求：
ZooKeeperServer實作了介面SessionExpirer，

// org.apache.zookeeper.server.ZooKeeperServer#expire
public void expire(Session session) {
    long sessionId = session.getSessionId();
    LOG.info(
        "Expiring session 0x{}, timeout of {}ms exceeded",
        Long.toHexString(sessionId),
        session.getTimeout());
    close(sessionId);
}

// org.apache.zookeeper.server.ZooKeeperServer#close
private void close(long sessionId) {
    Request si = new Request(null, sessionId, 0, OpCode.closeSession, null, null);
    submitRequest(si);
}

3、收集會話相關的臨時節點串列
在ZooKeeper中，一旦某個會話失效后，那么和該會話相關的臨時（EPHEMERAL）節點都需要被清理掉，因此，在清理臨時節點之前，首先需要將服務器上所有和該會話相關的臨時節點都整理出來，

在 ZooKeeper 的記憶體資料庫中，為每個會話都單獨保存了一份由該會話維護的所有臨時節點集合ephemerals，key為 sessionID ，value為臨時節點path集合，

Map<Long, HashSet<String>> ephemerals = new ConcurrentHashMap<Long, HashSet<String>>();

因此在會話清理階段，只需要根據當前即將關閉的會話的 sessionID 從記憶體資料庫中獲取到這份臨時節點串列 ephemerals 即可，

發起的“會話關閉”請求，流轉到 PrepRequestProcessor，由 PrepRequestProcessor做臨時節點收集作業，

在處理會話關閉請求時，可能正好有節點洗掉和節點創建的請求在進行：

節點洗掉請求，洗掉的目標節點正好是上述臨時節點中的一個，需要將其從上述獲取的臨時節點串列中排除，以免重復洗掉，
臨時節點創建請求，創建的目標節點正好是上述臨時節點中的一個，需要將其加入到上述獲取的臨時節點串列中，后續進行洗掉操作，

完成會話相關的臨時節點收集后，ZooKeeper會逐個將這些臨時節點轉換成“節點洗掉”請求，并放入事務變更佇列outstandingChanges中去，

同時將收集起來的ephemerals構建 CloseSessionTxn 設定給Request，

如下是 PrepRequestProcessor中處理會話關閉請求的部分代碼：

// org.apache.zookeeper.server.PrepRequestProcessor#pRequest2Txn
case OpCode.closeSession:
    synchronized (zks.outstandingChanges) {
        // 需要將 Ephemerals 臨時節點移動到 zks.outstandingChanges 中
        // 這個程序需要對 zks.outstandingChanges加同步鎖，否則將會出現：
        // 正在進行中的洗掉節點的事務，會重復執行洗掉操作，
        // need to move getEphemerals into zks.outstandingChanges
        // synchronized block, otherwise there will be a race
        // condition with the on flying deleteNode txn, and we'll
        // delete the node again here, which is not correct
        Set<String> es = zks.getZKDatabase().getEphemerals(request.sessionId);
        for (ChangeRecord c : zks.outstandingChanges) {
            if (c.stat == null) {
                // Doing a delete
                // stat = null，說明該節點正在進行洗掉操作，
                // 需要在 ephemerals 中將其排除
                es.remove(c.path);
            } else if (c.stat.getEphemeralOwner() == request.sessionId) {
                // outstandingChanges中有該會話的臨時節點，可能正在進行創建節點操作
                // 需要將其加入到 ephemerals中
                es.add(c.path);
            }
        }
        // 構建 節點洗掉 事務，并將其加入到 zks.outstandingChanges中
        for (String path2Delete : es) {
            if (digestEnabled) {
                parentPath = getParentPathAndValidate(path2Delete);
                parentRecord = getRecordForPath(parentPath);
                parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
                parentRecord.stat.setPzxid(request.getHdr().getZxid());
                parentRecord.precalculatedDigest = precalculateDigest(
                        DigestOpCode.UPDATE, parentPath, parentRecord.data, parentRecord.stat);
                addChangeRecord(parentRecord);
            }
            nodeRecord = new ChangeRecord(
                    request.getHdr().getZxid(), path2Delete, null, 0, null);
            nodeRecord.precalculatedDigest = precalculateDigest(
                    DigestOpCode.REMOVE, path2Delete);
            addChangeRecord(nodeRecord);
        }
        if (ZooKeeperServer.isCloseSessionTxnEnabled()) {
            // 將 ephemerals 構建 CloseSessionTxn 放進 request
            request.setTxn(new CloseSessionTxn(new ArrayList<String>(es)));
        }
        zks.sessionTracker.setSessionClosing(request.sessionId);
    }
    break;

4、移除會話：
會話關閉請求流轉到 FinalRequestProcessor，將其應用到記憶體資料庫，首先將會話從 SessionTracker 中清除，

5、洗掉臨時節點：
記憶體資料庫DataTree，從請求中取出CloseSessionTxn，決議出ephemerals，依次執行洗掉節點操作，

6、關閉 ServerCnxn：
最后，斷開與客戶端的連接，向網路底層發送 ServerCnxnFactory.closeConn（ByteBuffer.allocate(0)），NIOServerCnxn 在處理寫事件（NIOServerCnxn#handleWrite）時會觸發 CloseRequestException 例外，并捕獲該例外，進行連接斷開和清理作業，

會話激活

為了保持客戶端會話的有效性，在ZooKeeper的運行程序中，客戶端會在會話超時時間過期范圍內向服務端發送 PING 請求來保持會話的有效性，也就是心跳檢測，

如果客戶端發現在 sessionTimeout/3 時間內尚未和服務器進行過任何通信，即沒有向服務端發送任何請求，那么就會主動發起一個PING請求，服務端收到該請求后，重新激活對應的客戶端會話，這個重新激活的程序稱為 touchSession，實際上，只要客戶端有請求發送到服務端，就會觸發一次會話激活，

通過閱讀會話管理SessionTracker的原始碼，發現Learner（Follower+Observer）是不管理會話的，Leader才有管理會話的職責，而當客戶端連接的服務端是Learner時，向Learner發送心跳或者請求觸發的會話激活，還需要配合 Learner 和 Leader之間的心跳程序，

服務端接收客戶端的請求，在流轉鏈式RequestProcessor處理時，會先觸發會話激活 touchSession，對于Learner的LearnerSessionTracker來說， touchSession只是簡單的更新下touchTable，
當Learner接收到來自Leader的心跳，構建心跳回應時，將touchTable中的會話資訊（sessionId+sessionTimeout）全部取出來發送給Leader，讓Leader去觸發會話激活，

會話激活

// org.apache.zookeeper.server.SessionTrackerImpl#touchSession
public synchronized boolean touchSession(long sessionId, int timeout) {
    SessionImpl s = sessionsById.get(sessionId);
    // 檢查會話是否存在
    if (s == null) {
        logTraceTouchInvalidSession(sessionId, timeout);
        return false;
    }
    // 檢查會話是否正在關閉
    if (s.isClosing()) {
        logTraceTouchClosingSession(sessionId, timeout);
        return false;
    }
    // 更新會話
    updateSessionExpiry(s, timeout);
    return true;
}

會話激活的程序，就是會話從舊桶遷移到新桶的程序：

// org.apache.zookeeper.server.ExpiryQueue#update
public Long update(E elem, int timeout) {
    // 1.獲取會話實體上一次過期時間點
    Long prevExpiryTime = elemMap.get(elem);
    long now = Time.currentElapsedTime();
    // 2.計算的 newExpiryTime 比 now + timeout 要大
    Long newExpiryTime = roundToNextInterval(now + timeout);
    // 3. prevExpiryTime 和 newExpiryTime 進行比較，相等就不做處理
    if (newExpiryTime.equals(prevExpiryTime)) {
        // No change, so nothing to update
        return null;
    }
    // 4.將 會話實體放進新桶里
    // First add the elem to the new expiry time bucket in expiryMap.
    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);

    // 5. 將舊桶中的對應會話記錄洗掉
    // 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);
    if (prevExpiryTime != null && !newExpiryTime.equals(prevExpiryTime)) {
        Set<E> prevSet = expiryMap.get(prevExpiryTime);
        if (prevSet != null) {
            prevSet.remove(elem);
        }
    }
    return newExpiryTime;
}

LearnerSessionTracker可以開啟本地會話管理LocalSessionTracker，LocalSessionTracker繼承則SessionTrackerImpl，故而會話管理的原理是一樣的，默認是不開啟的，

如若文章有錯誤理解，歡迎批評指正，同時非常期待你的評論、點贊和收藏，

如果想了解更多優質文章，和我更密切的學習交流，請關注如下同名公眾號【徐同學呀】，期待你的加入，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/437038.html

標籤：其他

上一篇：MapReduce on HBase stu表

下一篇：zookeeper3.5.9leader選舉原始碼分析類圖模型