使用過zookeeper的都知道，當(dāng)我們使用zookeeper創(chuàng)建一個節(jié)點時，我們能選擇節(jié)點的類型是“臨時節(jié)點”還是“永久節(jié)點”。臨時節(jié)點和永久節(jié)點的區(qū)別是，臨時節(jié)點會在客戶端斷開連接時被刪除，而永久節(jié)點無論客戶端是否斷開連接，都會保留。

臨時節(jié)點非常重要，我們經(jīng)常利用它來實現(xiàn)分布式鎖、選舉等等

而為了實現(xiàn)臨時節(jié)點的功能，zookeeper服務(wù)端就勢必要有一套高效的session管理機制，它能實現(xiàn)如下功能：當(dāng)客戶端session失效后，服務(wù)端能感知到，隨后刪掉客戶端當(dāng)前創(chuàng)建的臨時節(jié)點，并通知給其他的客戶端。這篇文章會深入探討zookeeper的session管理機制。

zookeeper的心跳機制

為什么要有心跳機制

zookeeper底層支持兩種網(wǎng)絡(luò)庫，一種是zookeeper基于NIO自己寫的，一種是Netty。那么zookeeper能不能直接通過感知TCP連接是否斷開來感知客戶端連接是否斷開呢？

答案是不能，原因有很多，個人覺得最重要的一點是，基于TCP連接來判斷客戶端是否存活是不靠譜的。

這里舉兩個異常case 1、客戶端進程直接crash了，還沒來得及發(fā)送FIN報文，這種情況下，zookeeper在TCP這一側(cè)是沒辦法及時感知到TCP連接已經(jīng)失效了。（也就是說，由于zookeeper沒收到client的FIN包，雖然client已經(jīng)掛了，TCP側(cè)還認(rèn)為客戶端還活著。只能等弱雞keepalive了）

2、客戶端和服務(wù)端之間很久沒有報文交互，TCP連接其實已經(jīng)失效了。（失效原因很多，比如路由器出問題了，網(wǎng)絡(luò)設(shè)備故障了等等）這時候，如果客戶端發(fā)送報文給服務(wù)端，linux會進行重試，默認(rèn)差不多要重試15分鐘，才能感知到這個連接已經(jīng)失效。

這里只是舉個例子。實際來說，client和zookeeper的報文交互可能不會那么少。也就是第二種情況，客戶端很久不給服務(wù)端發(fā)報文，要發(fā)了，才發(fā)現(xiàn)tcp連接已經(jīng)有問題了，這種情況不太可能出現(xiàn)。（當(dāng)然，選舉這種場景是有可能出現(xiàn)這種情況的，后面會單獨寫文章分析這個case） 但是從上面兩個例子，我們也不難得出結(jié)論，通過tcp連接的斷開與否來感知客戶端是否存活，似乎并不太靠譜。

client發(fā)送心跳

因為tcp的“不靠譜”，zookeeper為了能夠?qū)崿F(xiàn)可靠的連接管理（也為了保活），選擇自己實現(xiàn)心跳機制。

正如上圖所示，client隔一段時間就會發(fā)送一個心跳報文給服務(wù)端，告訴zookeeper自己還活著，別把我連接關(guān)了。

注意：這里亂入了一個create報文，因為正常報文也算是一種“心跳”。反正，zookeeper只要能收到報文，就能知道客戶端還活著。

服務(wù)端收到報文后，會更新session信息

這里帶大家看看代碼。很簡單，收到報文后，會調(diào)用SessionTracker.touchSession()來更新session信息

sessionId是zookeeper為每一個連接分配的唯一id

管理session的難點在哪

zookeeper 管理session的需求分析

這里我們思考下zookeeper對于session管理的需求是什么？

1、zookeeper需要有個地方存session

2、當(dāng)客戶端一段時間沒發(fā)生心跳時，zookeeper要能感知到

第一個問題比較簡單，java提供了各式各樣的集合，無論是Map或者是List理論上都能存session。由于我們一般是通過sessionId來找到關(guān)聯(lián)的Session的，因此使用Map更合適點。Zookeeper也是這么做的protected final ConcurrentHashMap sessionsById這里sessionsById 就是zookeeper管理Session使用的容器

第二個問題看起來也很簡單，客戶端不是會發(fā)送心跳么？我給每一個session記錄一個上一個報文到達時間，一旦收到新的報文，我就更新這個時間。然后我再不斷地掃描每個session是否很久沒收到報文不就行了？

如何檢測Session失效？

按照上面的說法，收到一個報文，我就更新Session的“上次報文”字段。假設(shè)session失效時間是4秒，我就每隔4秒掃描一次session的集合，找出那個超過4秒沒有新的報文的session不就行了？像上圖一樣，假設(shè)sessionTimeout是4秒，現(xiàn)在已經(jīng)11點50分05秒了，理論上每個session上一個報文時間應(yīng)該大于11點50分01秒。很明顯，session1失效了。

定時任務(wù)的選型

既然要定時掃描，我們就需要跑一個定時任務(wù)。jdk本身也提供了很多的定時任務(wù)方案。不知道定時任務(wù)的同學(xué)可以參考這篇文章Java中定時任務(wù)的6種實現(xiàn)方式，你知道幾種？- 掘金既然如此，我們完全可以使用jdk自帶的定時任務(wù)，定時去掃描這個集合啊，這樣不就能很輕易的找到失效的session了么？

這里有兩個問題：1、每一個客戶端和服務(wù)端的連接，sessionTimeout都是可配的。我們例子是4秒沒收到報文，就認(rèn)為連接失效。實際超時時間可能有1秒、2秒、3秒.... 所以如果用定時任務(wù)來實現(xiàn)，我們可能需要啟動不止一個定時任務(wù)。2、jdk提供的定時任務(wù)不夠靈活，什么意思呢。比如我設(shè)置的sessionTimeout是4秒，現(xiàn)在是11點。然后我在11點00分02秒就收到了一個心跳，那么下次檢測時間應(yīng)該變成11點00分06秒。而jdk的定時任務(wù)限定了，只能每隔4秒檢測一次。比如：11點、11點00分04秒、11點00分08秒、這樣檢測下去。而如果用jdk的定時任務(wù)，我們只能簡單的隔一段時間，檢測一次。這里可以仔細體會下兩者的差異。

想一想這些問題，是不是發(fā)現(xiàn)想實現(xiàn)一個高效的session管理機制是不是沒那么簡單。接下來我們看看，zookeeper是如何巧妙地實現(xiàn)session的管理。

zookeeper session管理機制

接下來就要介紹zookeeper的session管理機制了。

1、通過expiryMap存儲過期時間與session集合的對應(yīng)關(guān)系

首先，zookeeper內(nèi)部有一個expiryMap

非常簡單，Key是過期時間，value是一個Set，里面放了一個個的Session。

舉個例子：S1在11點50分02秒的key下面，表示如果11點50分02秒前沒收到新的報文，就認(rèn)為S1過期了。

2、當(dāng)收到某條連接的報文時，更新expiryMap

拿上圖的S1為例，它的sessionTimeout是4秒，在11點50分01秒收到報文。那么理論上下個session檢測時間會是 11點50分05秒。

這里要說下expiryMap的第一個特征，它的key并不是隨意一個時間。它會間隔一個固定的時間叫做expirationInterval，數(shù)值上它等于zookeeper的配置tickTime（默認(rèn)配的2秒）

所以說，這里計算出11點50分05秒后，它會round下，round到11點50分06秒

如圖所示：

S1在截止時間前更新了session，我們就要把它從舊的桶里移除，挪到新的桶里。

3、循環(huán)檢測ExpiryMap

有個SessionTracker線程會循環(huán)檢測這個expiryMap，找到最近的那個key對應(yīng)的session集合，把他們?nèi)慷歼^期掉。

就像上面的例子，一旦時間到了11點50分02秒，就把對應(yīng)的session全部過期掉。

小結(jié)下

1、如果收到報文，會把session放到下一個過期桶里。2、SessionTracker會按次序，不斷地取出過期的桶，把桶里的session全部過期掉（過期會刪除臨時節(jié)點，當(dāng)然還有其他一系列操作） 3、zookeeper底層使用了非常簡單的Map就實現(xiàn)了非常高效的Session管理機制。

session管理機制源碼分析

接下來我們來看看源碼

1、server端的心跳續(xù)約

//org.apache.zookeeper.server.ExpiryQueue#update
public Long update(E elem, int timeout) {
  //1、除了上面我們介紹的ExpiryMap，zookeeper內(nèi)部還有一個elemMap，用于存放 Session -> 過期時間
  Long prevExpiryTime = elemMap.get(elem);
  long now = Time.currentElapsedTime();
  //2、收到心跳后，我們會計算session應(yīng)該更新到哪個桶里
  Long newExpiryTime = roundToNextInterval(now + timeout);


  //桶不變，就不用更新expiryMap了
  if (newExpiryTime.equals(prevExpiryTime)) {
    // No change, so nothing to update
    return null;
  }


  // First add the elem to the new expiry time bucket in expiryMap.
  //3. 找到新的桶，插入進去
  Set set = expiryMap.get(newExpiryTime);
  if (set == null) {
    // Construct a ConcurrentHashSet using a ConcurrentHashMap
    set = Collections.newSetFromMap(new ConcurrentHashMap());
    // Put the new set in the map, but only if another thread
    // hasn't beaten us to it
    Set existingSet = expiryMap.putIfAbsent(newExpiryTime, set);
    if (existingSet != null) {
      set = existingSet;
    }
  }
  set.add(elem);


  // 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);
  //4. 從舊的桶里移除
  if (prevExpiryTime != null && !newExpiryTime.equals(prevExpiryTime)) {
    Set prevSet = expiryMap.get(prevExpiryTime);
    if (prevSet != null) {
      prevSet.remove(elem);
    }
  }
  return newExpiryTime;
}

其實就是這幅圖 1、放入一個：Session -> 過期時間 的Map中 2、收到心跳后，我們會計算session應(yīng)該更新到哪個桶里 3、找到新的桶，插入進去 4、把session從舊的桶里移除

2、SessionTracker不斷地輪訓(xùn)，找到過期的Session集合，然后都過期掉

//org.apache.zookeeper.server.SessionTrackerImpl#run
@Override
public void run() {
  try {
    while (running) {
      //1. 這個其實就是不斷地輪訓(xùn)下一個要檢測的key
      // 比如按我們的例子，應(yīng)該是11點50分02秒檢測一次、11點50分04秒檢測一次、11點50分06秒檢測一次...
      // 這里的waitTime就是找到下次檢測需要等待多久，比如現(xiàn)在是11點50分01秒了，這個waitTime就是1秒
      // 如果是11點50分02秒了，waitTime就是0，我們要開始把過期的session都失效掉了
      long waitTime = sessionExpiryQueue.getWaitTime();
      if (waitTime > 0) {
        Thread.sleep(waitTime);
        continue;
      }


      //2. 取出過期的Session集合，全部都expire掉，如果session都及時發(fā)送了心跳了，這里就會拿到一個空的集合
      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!");
}

其實就是這個圖