Vert.x系列（四）-- HAManager源碼分析

三．HA模式。

前言：在集羣模式下，Vert.x框架擁有HA（HighAvailability）能力。通俗的解釋是集羣中的一個節點跪了，原來運行在失敗節點上的Verticle會在其餘節點上自動啓動。固然，前提是Verticle被設置了HA模式。

原理：
首先，HA的構造方法裏傳入了ClusterManager ,利用這個類實現node的add/left 監聽，
這樣集羣的每一個節點得到其餘節點的add/left。每當add事件發生，把節點的信息存到集羣貢獻的map中，除非集羣崩潰，信息都回在。每當left事件發生，存活的節點肯定本身是否與失敗節點在同一個組中：
若是不是同一個組，那麼它將不會做爲故障轉移節點的候選者。羣集中的節點僅故障轉移到同一組中的其餘節點；
若是是同一個組，那麼用失敗節點的UUID計算Hashcode，而後用這個Hashcode對集羣節點的個數作取模(%)運算，以取模的結果做爲索引從集羣全部節點的List拿到對應的節點，判斷這個計算的出來的節點是否是自身。若是是自身就re-deploy那個失敗的節點。

其中有個重要的邏輯檢查一個「集羣最小節點數」quorum的值。（quorum本意：會議法定最小參加人數）。官方說能夠避免quorum 丟失後致使的競爭條件，若是不檢查就要使用排他鎖，而排他鎖太棘手，很容易致使死鎖。（原文： avoid race conditions resulting in modules being deployed after a quorum has been lost，and without having to resort to exclusive locking which is actually quite tricky here, and prone to deadlock。）
讓我來猜想、估計、YY這段話的原意：框架實現了最小quorum臺服務器存活才稱爲集羣，才實現HA模式。即 if(aliveServerNum > quorum) {
//  doHA
}
可是這個aliveServerNum 必然分別維護在這N臺服務上。如何維護這個aliveServerNum？
一臺服務器跪了，須要aliveServerNum -- 。一臺服務器加了，須要aliveServerNum++。若是有多臺服務器同時加入，代碼須要設計爲
// get lock 
{
 aliveServerNum++
} 
// release lock
那麼，能夠想象的場景：S1,S2服務器發生重啓,要各自通知到
A1, A2,A3...An 個alive的服務器。
S1對 A1 的aliveServerNum++鎖定，請求 A2的鎖。
S2對 A2 的aliveServerNum++鎖定，請求 A1的鎖。 

與其費工夫維護aliveServerNum，還不如用 checkQuorum（）方法和boolean attainedQuorum代替。即
checkQuorum()
if(attainedQuorum) {
//  doHA
}
可是，這裏我有個疑問是：
若是lock是集羣鎖纔有上面的死鎖問題，若是lock是單機鎖synchronize，好像能解決問題又不會有死鎖？但不管如何，不維護aliveServerNum是更明智的選擇。

代碼：
private final VertxInternal vertx;   // vertx
private final DeploymentManager deploymentManager;  // deploymentManager
private final ClusterManager clusterManager;  //  接口，重要方法是 join leave nodeListener

private final int quorumSize;          // 集羣最小節點數
private final String group;           // 相同group纔有HA邏輯
private final JsonObject haInfo;                       
private final Map<String, String> clusterMap; //
// 別的屬性是單機的，當服務器跪掉就丟失了
// 爲了failover時還能取得當時的信息。須要把信息變成
// 集羣共享，在須要時用clusterManager.getSyncMap()取得。 haInfo和clusterMap就是用來作這件事
// clusterMap 的存在乎義是方便用nodeID查找。

private final String nodeID;      // 節點ID，惟一性。
private final Queue<Runnable> toDeployOnQuorum = new ConcurrentLinkedQueue<>();
// 若是節點跪了時，集羣不在 quorum 狀態。 那麼就把信息放在這個隊列容器中。而後用定時器（vertx.setPeriodic）不停的檢測。 等達到attainedQuorum再deploy.

private final boolean enabled;  // 是否HA的總開關。

private long quorumTimerID;  //  見上面 toDeployOnQuorum 。定時器的ID，達到條件後調用vertx.cancelTimer(quorumTimerID);取消

private volatile boolean attainedQuorum; // 見 「原理」
private volatile FailoverCompleteHandler failoverCompleteHandler; // 提供自定義的重啓邏輯插入
private volatile boolean failDuringFailover; // For testing:
private volatile boolean stopped; // 狀態量
private volatile boolean killed; // 狀態量
private Consumer<Set<String>> clusterViewChangedHandler;

HAManager 的構造方法

clusterManager.nodeListener(new NodeListener() {

  @Override
  public void nodeAdded(String nodeID) {
    HAManager.this.nodeAdded(nodeID);
  }

  @Override
  public void nodeLeft(String leftNodeID) {
    HAManager.this.nodeLeft(leftNodeID);
  }
});

存在 nodeAdded 和 nodeLeft的監聽。clusterManager 是接口，具體還要看由實現類。

private void doDeployVerticle(final String verticleName, DeploymentOptions deploymentOptions,
                              final Handler<AsyncResult<String>> doneHandler) {
  final Handler<AsyncResult<String>> wrappedHandler = asyncResult -> {
    if (asyncResult.succeeded()) {
      // Tell the other nodes of the cluster about the verticle for HA purposes
      addToHA(asyncResult.result(), verticleName, deploymentOptions);
    }
    if (doneHandler != null) {
      doneHandler.handle(asyncResult);
    } else if (asyncResult.failed()) {
      log.error("Failed to deploy verticle", asyncResult.cause());
    }
  };
  deploymentManager.deployVerticle(verticleName, deploymentOptions, wrappedHandler);
}

這個方法一開始看了讓人蒙圈，沒有搞明白前面的asyncResult怎麼就succeeded了。其實，方法前部全部的代碼都是爲了定義一個wrappedHandler ，定義完了在方法最後一句傳到deploymentManager.deployVerticle裏。跟蹤代碼，在若干方法棧後纔有對這個wrappedHandler.handle()方法調用:

completionHandler.handle(result);

因此，額外說句，這個方法的名字其實有問題。通常名爲doX的方法，都是處理最底層的X業務，而這裏實際上是交付給了deploymentManager去作。

private String chooseHashedNode(String group, int hashCode)

這個方法會計算出從新部署的節點，而後每一個存活的節點都調用，判斷結果是否是節點本身。

String chosen = chooseHashedNode(group, failedNodeID.hashCode());
if (chosen != null && chosen.equals(this.nodeID)) {

若是是本身，才繼續恢復流程，調用核心業務方法。若是不是本身，沒有事情發生。

private void processFailover(JsonObject failedVerticle)

processFailover()使用了 CountDownLatch類 處理線程阻塞和超時處理邏輯。

別的方法都很簡單的，看名字就知道內容。