ZooKeeper的選舉過程預設使用FastLeaderElection類，FastLeaderElection啟動時啟動Messenger收發選舉資訊。選舉完成後選出1個Leader和若干Follower。

首先理解幾個概念：   

Epoch：投票週期，用於區分每一個round，每一次建立一個新的leader-follower關係，都會有一個唯一的epoch值去標識。就好像皇帝登基必須得有一個年號，與之前或之後的皇帝進行區分。剛啟動是從檔案讀取儲存的currentEpoch和acceptedEpoch值。預設是-1。

zxid：事務ID，表示Zookeeper當前寫操作的序列號，確保寫操作的按順序執行。有一種場景是當Follower的最大zxid大於Leader的zxid時，Leader會發送TRUNC包給Follower截斷多餘的事務，保證和Leader資料一致。zxid會在ZxDataBase的loadDataBase方法中初始化。每次執行一次寫事務zxid就會加1，預設為0。

選舉規則：先判斷雙方的Epoch，留取大的Epoch；再看雙方的zxid，留取大的zxid一方。最後的結果就是（Epoch<32|zxid）的一方會被推舉為Leader。

叢集間選舉使用Messenger來負責選舉資訊的互動，Messenger底層使用QuorumCnxManager管理本機和叢集其他機器之間的Socket資料傳遞，通訊流程示意圖如下所示。

叢集中所有LOOKING狀態的機器同時啟動選舉過程。選舉結束則產生一個Leader和多個Follower。然後就是各Follower連線到Leader進行事務初始化，實現資料同步。

Observer不參與選舉，但會接收Leader的資訊同步要求。

QuorumPeer剛啟動時會設定初始狀態為LOOKING，然後啟動FastLeaderElection選舉過程，首先將自己作為Vote群發給其他QuorumPeer，群發的資料結構為Notification。同時啟動Messenage的收發執行緒，呼叫lookForLeader方法發起一輪選舉Leader的過程。

在Leader/Follower狀態，如果接收到遠方LOOKING節點尋找Leader而傳送的Notification包，則Messenger會自動回覆Leader資訊給對方；在LOOKING狀態，則Messenger會回覆當前Vote節點（臨時Leader）資訊給對方。

預設選舉規則規定當超過半數的QuorumPeer都認同同一個Leader時，選舉過程結束，各QuorumPeer將自己設定為Leader、Follower或Observer。當然有興趣的讀者也可以嘗試建立自己的選舉規則類，只要實現QuorumVerifier即可。

進行Fast leader election的先決條件：

1、 每個QuorumPeer都知道其他QuorumPeer的ip地址，並知道QuorumPeer的總數。

2、 每個QuorumPeer一開始都是發起一個vote，選取自己為leader。向其他所有的QuorumPeer傳送vote的notification，並等待回覆。

3、 根據QuorumPeer的狀態處理vote notification訊息。

選舉過程中產生的臨時Leader成為Vote，當選舉結束後最後一輪產生的Vote即成為新的Leader。

QuorumPeer每次發起選舉都呼叫lookForLeader方法實現，首先將自己設定為LOOKING狀態。該方法是FastLeaderElection類的主方法，具體的流程如下：

• 首先更新選舉週期logicalclock， 並把自己作為leader作為投票發給所有其他的server。
• 然後進入本輪投票的迴圈，直到自己不再是LOOKING狀態。

1. 從recvqueue獲取一個網路包（recvqueue的資料來自Messenger），如果沒有收到包則檢查是否要重連和重發自己的投票。
2. 收到投票後判斷對方投票的狀態。

1. LOOKING：
   • 如果對方投票的週期（Epoch）大於自己的週期（Epoch），那就清空自己的已經收到的投票集合recvset，並將自己作為候選和對方投票的leader做比較，選出大的作為新的投票，然後再發送給所有人。 這裡比較大小是通過比較（zxid，sid）這個二元組來的，zxid大的就大，否則sid大的就大。
   • 如果對方的投票週期小於自己，則忽略對方的投票。
   • 如果週期相等，則比較對方的投票和自己認為的候選，選出大的作為新的候選，然後再發送給所有人。
   • 然後判斷當前收到的投票是否可以得出誰是leader的結論，這裡主要是通過判斷當前的候選leader在收到的投票中是否佔了多數。
   • 如果候選leader在收到的投票中佔了多數，則再等待finalizeWait時鐘，看是否有人修改leader的候選，如果修改了則把投票放到recvqueue中再重新迴圈。
2. OBSERVING：如果對方是一個觀察者，由於它沒有投票權，則無視它
3. FOLLOWING或LEADING：

• 如果對方和自己再一個時鐘週期，說明對方已經完成選舉，如果對方說它是leader，那我們就把它作為leader，否則就要比較下對方選舉的leader在自己這裡是否佔有多數，並且選舉的leader確認了願意當leader，如果都通過了，就把這個候選作為自己的leader
• 如果對方和自己不在一個時鐘週期，說明自己掛掉後又恢復起來，這個時候把別人的投票收集到一個單獨的集合outofelection（從名字可以看出這個集合不是用在選舉判斷），如果對方的投票在outofelection中佔有大多數，並且leader也確認了自己願意做leader，這個時候更新自己的選舉週期logicalclock，並修改自己的狀態為FOLLOWING或LEADING

 

lookForLeader程式碼較長，我們先看看它的主體結構。

public Vote lookForLeader() throws InterruptedException {
     try {
        HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();
        HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();
       int notTimeout = finalizeWait;
        synchronized(this){
             logicalclock.incrementAndGet();
             updateProposal(getInitId(), getInitLastLoggedZxid(), 
getPeerEpoch());
       }
        sendNotifications();
        while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){           
            Notification n = recvqueue.poll(notTimeout,
                        TimeUnit.MILLISECONDS);
            if (self.getCurrentAndNextConfigVoters().contains(n.sid)) {
               switch (n.state) {
                  case LOOKING:
                       {A程式碼}
                        break;
                    case OBSERVING:
                        LOG.debug("Notification from observer: " + n.sid);
                        break;
                    case FOLLOWING:
                    case LEADING:                    
                        {B程式碼}
                        break;
                    default:
                        LOG.warn("Notification state unrecoginized: " + n.state
                              + " (n.state), " + n.sid + " (n.sid)");
                        break;
                    }
                } 
            }
            return null;
        }   
 }

具體分析如下：

首先通過sendNotification方法告訴叢集我在尋找Leader，叢集中其他機器會在Messenger程序中接收到Notification，然後都會回覆誰應該是當前Leader。

FastLeaderElection收集到足夠的Notification訊息，來判斷到底誰才是合法的Leader。為此它對每條Notification訊息進行下列判斷：

A. 回覆Notification的傳送方也是LOOKING狀態

如果回覆Notification的傳送方也是LOOKING狀態，說明它還不知道最終的Leader是誰，這時候FastLeaderElection和傳送方比較，看看誰的潛在Leader的Epoch和zxid最大，最大的設定為當前的候選Leader，然後將這個候選Leader廣播出去，讓傳送方也能更改自己的後續Leader。同時判斷自己的收集的Notification能否達到超過半數的條件從而決定最終的Leader，如果能決定則設定最終Leader，退出選舉過程。

case LOOKING:   
   // If notification > current, replace and send messages out
   if (n.electionEpoch > logicalclock.get()) {
       logicalclock.set(n.electionEpoch);
       recvset.clear();
       if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                       getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
            updateProposal(n.leader, n.zxid, n.peerEpoch);
       } else {
         updateProposal(getInitId(),getInitLastLoggedZxid(),getPeerEpoch());
       }
       sendNotifications();
   } else if (n.electionEpoch < logicalclock.get()) {
       break;
   } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,proposedLeader, proposedZxid, proposedEpoch)) {
       updateProposal(n.leader, n.zxid, n.peerEpoch); sendNotifications();
   }
   recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, 
n.peerEpoch));
   if (termPredicate(recvset, new Vote(proposedLeader, proposedZxid,
                            logicalclock.get(), proposedEpoch))) {
      while((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null){
         if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,proposedLeader, proposedZxid, proposedEpoch)){
             recvqueue.put(n);
             break;
         }
      }
      if (n == null) {
           self.setPeerState((proposedLeader == self.getId()) ?
                                        ServerState.LEADING: learningState());
           Vote endVote = new Vote(proposedLeader, proposedZxid, proposedEpoch);
           leaveInstance(endVote);
        return endVote;
      }
  }
  break;

totalOrderPredicate方法用於判斷對方傳送過來的Vote是不是更新的Leader候選者,如果是的話則更新本地proposedLeader。主要程式碼如下：

protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
        return ((newEpoch > curEpoch) ||
                ((newEpoch == curEpoch) &&  ((newZxid > curZxid) ||
 ((newZxid == curZxid) && (newId > curId)))));
}

termPredicate用於判斷是否滿足選出Leader條件，QuorumVerifier介面的實現，如果滿足則設定Leader，並退出FastLeaderElection過程。主要程式碼如下：

private boolean termPredicate(HashMap<Long, Vote> votes, Vote vote) {
        SyncedLearnerTracker voteSet = new SyncedLearnerTracker();
        voteSet.addQuorumVerifier(self.getQuorumVerifier());
        if (self.getLastSeenQuorumVerifier() != null
                && self.getLastSeenQuorumVerifier().getVersion() > self
                        .getQuorumVerifier().getVersion()) {
            voteSet.addQuorumVerifier(self.getLastSeenQuorumVerifier());
        }
        for (Map.Entry<Long, Vote> entry : votes.entrySet()) {
            if (vote.equals(entry.getValue())) {
                voteSet.addAck(entry.getKey());
            }
        }
        return voteSet.hasAllQuorums();
}

 

B. 回覆Notification的傳送方也是Leader或Follower

如果回覆Notification的傳送方是Leader或者Follower，則流程比較簡單，將Notification訊息儲存到recvset中，並呼叫termPredicate方法判斷是否能確定Leader並結束選舉過程。

程式碼片段：

case FOLLOWING:
case LEADING:
     if(n.electionEpoch == logicalclock.get()){
         recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, 
n.peerEpoch));
         if(termPredicate(recvset, new Vote(n.leader, n.zxid, n.electionEpoch,
n.peerEpoch, n.state))  && checkLeader(outofelection, n.leader, 
n.electionEpoch)) {
            self.setPeerState((n.leader == self.getId()) ?
                                  ServerState.LEADING: learningState());
            Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
            leaveInstance(endVote);
            return endVote;
         }
    }

相關推薦