Zookeeper-Zookeeper leader選舉

在上一篇文章中咱們大體瀏覽了zookeeper的啓動過程，而且提到在Zookeeper的啓動過程當中leader選舉是很是重要並且最複雜的一個環節。那麼什麼是leader選舉呢？zookeeper爲何須要leader選舉呢？zookeeper的leader選舉的過程又是什麼樣子的？本文的目的就是解決這三個問題。

首先咱們來看看什麼是leader選舉。其實這個很好理解，leader選舉就像總統選舉同樣，每人一票，得到多數票的人就當選爲總統了。在zookeeper集羣中也是同樣，每一個節點都會投票，若是某個節點得到超過半數以上的節點的投票，則該節點就是leader節點了。

國家選舉總統是爲了選一個最高統帥，治理國家。那麼zookeeper集羣選舉的目的又是什麼呢？其實這個要清楚明白的解釋仍是挺複雜的。咱們能夠簡單點想這個問題：咱們有一個zookeeper集羣，有好幾個節點。每一個節點均可以接收請求，處理請求。那麼，若是這個時候分別有兩個客戶端向兩個節點發起請求，請求的內容是修改同一個數據。好比客戶端c1，請求節點n1，請求是set a = 1; 而客戶端c2，請求節點n2，請求內容是set a = 2;

那麼最後a是等於1仍是等於2呢？ 這在一個分佈式環境裏是很難肯定的。解決這個問題有不少辦法，而zookeeper的辦法是，咱們選一個總統出來，全部的這類決策都提交給總統一我的決策，那以前的問題不就沒有了麼。

那咱們如今的問題就是怎麼來選擇這個總統呢？ 在現實中，選擇總統是須要宣講拉選票的，那麼在zookeeper的世界裏這又如何處理呢？咱們仍是show code吧。

在QuorumPeer的startLeaderElection方法裏包含leader選舉的邏輯。Zookeeper默認提供了4種選舉方式，默認是第4種: FastLeaderElection。

咱們先假設咱們這是一個嶄新的集羣，嶄新的集羣的選舉和以前運行過一段時間的選舉是有稍許不一樣的，後面會說起。

節點狀態： 每一個集羣中的節點都有一個狀態 LOOKING, FOLLOWING, LEADING, OBSERVING。都屬於這4種，每一個節點啓動的時候都是LOOKING狀態，若是這個節點參與選舉但最後不是leader，則狀態是FOLLOWING，若是不參與選舉則是OBSERVING，leader的狀態是LEADING。

開始這個選舉算法前，每一個節點都會在zoo.cfg上指定的監聽端口啓動監聽(server.1=127.0.0.1:20881:20882)，這裏的20882就是這裏用於選舉的端口。

在FastLeaderElection裏有一個Manager的內部類，這個類裏有啓動了兩個線程：WorkerReceiver， WorkerSender。爲何說選舉這部分複雜呢，我以爲就是這些線程就像左右互搏同樣，很是難以理解。顧名思義，這兩個線程一個是處理從別的節點接收消息的，一個是向外發送消息的。對於外面的邏輯接收和發送的邏輯都是異步的。

這裏配置好了，QuorumPeer的run方法就開始執行了，這裏實現的是一個簡單的狀態機。由於如今是LOOKING狀態，因此進入LOOKING的分支，調用選舉算法開始選舉了：

setCurrentVote(makeLEStrategy().lookForLeader());

而在lookForLeader裏主要是幹什麼呢？首先咱們會更新一下一個叫邏輯時鐘的東西，這也是在分佈式算法裏很重要的一個概念，可是在這裏先不介紹，能夠參考後面的論文。而後決定我要投票給誰。不過zookeeper這裏的選舉真直白，每一個節點都選本身(汗),選我，選我，選我...... 而後向其餘節點廣播這個選舉信息。這裏實際上並無真正的發送出去，只是將選舉信息放到由WorkerSender管理的一個隊列裏。

synchronized(this){
    //邏輯時鐘           
    logicalclock++;
    //getInitLastLoggedZxid(), getPeerEpoch()這裏先不關心是什麼，後面會討論
    updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
}

//getInitId() 便是獲取選誰，id就是myid裏指定的那個數字，因此說必定要惟一
private long getInitId(){
        if(self.getQuorumVerifier().getVotingMembers().containsKey(self.getId()))       
            return self.getId();
        else return Long.MIN_VALUE;
}

//發送選舉信息，異步發送
sendNotifications();

如今咱們去看看怎麼把投票信息投遞出去。這個邏輯在WorkerSender裏，WorkerSender從sendqueue裏取出投票，而後交給QuorumCnxManager發送。由於前面發送投票信息的時候是向集羣全部節點發送，因此固然也包括本身這個節點，因此QuorumCnxManager的發送邏輯裏會判斷，若是這個要發送的投票信息是發送給本身的，則不發送了，直接進入接收隊列。

public void toSend(Long sid, ByteBuffer b) {
        if (self.getId() == sid) {
             b.position(0);
             addToRecvQueue(new Message(b.duplicate(), sid));
        } else {
             //發送給別的節點，判斷以前是否是發送過
             if (!queueSendMap.containsKey(sid)) {
                 //這個SEND_CAPACITY的大小是1，因此若是以前已經有一個還在等待發送，則會把以前的一個刪除掉，發送新的
                 ArrayBlockingQueue<ByteBuffer> bq = new ArrayBlockingQueue<ByteBuffer>(SEND_CAPACITY);
                 queueSendMap.put(sid, bq);
                 addToSendQueue(bq, b);

             } else {
                 ArrayBlockingQueue<ByteBuffer> bq = queueSendMap.get(sid);
                 if(bq != null){
                     addToSendQueue(bq, b);
                 } else {
                     LOG.error("No queue for server " + sid);
                 }
             }
             //這裏是真正的發送邏輯了
             connectOne(sid);
                
        }
    }

connectOne就是真正發送了。在發送以前會先把本身的id和選舉地址發送過去。而後判斷要發送節點的id是否是比本身的id大，若是大則不發送了。若是要發送又是啓動兩個線程：SendWorker,RecvWorker(這種一個進程內許多不一樣種類的線程，各自幹活的狀態真的很難理解)。發送邏輯還算簡單，就是從剛纔放到那個queueSendMap裏取出，而後發送。而且發送的時候將發送出去的東西放到一個lastMessageSent的map裏，若是queueSendMap裏是空的，就發送lastMessageSent裏的東西，確保對方必定收到了。

看完了SendWorker的邏輯，再來看看數據接收的邏輯吧。還記得前面提到的有個Listener在選舉端口上啓動了監聽麼，如今這裏應該接收到數據了。咱們能夠看到receiveConnection方法。在這裏，若是接收到的的信息裏的id比自身的id小，則斷開鏈接，並嘗試發送消息給這個id對應的節點(固然，若是已經有SendWorker在往這個節點發送數據，則不用了)。

若是接收到的消息的id比當前的大，則會有RecvWorker接收數據，RecvWorker會將接收到的數據放到recvQueue裏。

而FastLeaderElection的WorkerReceiver線程裏會不斷地從這個recvQueue裏讀取Message處理。在WorkerReceiver會處理一些協議上的事情，好比消息格式等。除此以外還會看看接收到的消息是否是來自投票成員。若是是投票成員，則會看看這個消息裏的狀態，若是是LOOKING狀態而且當前的邏輯時鐘比投票消息裏的邏輯時鐘要高，則會發個通知過去，告訴誰是leader。在這裏，剛剛啓動的嶄新集羣，因此邏輯時鐘基本上都是相同的，因此這裏還沒判斷出誰是leader。不過在這裏咱們注意到若是當前節點的狀態是LOOKING的話，接收邏輯會將接收到的消息放到FastLeaderElection的recvqueue裏。而在FastLeaderElection會從這個recvqueue裏讀取東西。

這裏就是選舉的主要邏輯了：totalOrderPredicate

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

1. 判斷消息裏的epoch是否是比當前的大，若是大則消息裏id對應的server我就認可它是leader

2. 若是epoch相等則判斷zxid，若是消息裏的zxid比個人大我就認可它是leader

3. 若是前面兩個都相等那就比較一下server id吧，若是比個人大我就認可它是leader。

關於前面兩個東西暫時咱們不去關心它，對於新啓動的集羣這二者都是相等的。

那這樣看來server id的大小也是leader選舉的一環啊（有的人生下來註定就不平凡，這都是命啊）。

最後咱們來看看，不少文章所介紹的，若是超過一半的人說它是leader，那它就是leader的邏輯吧

private boolean termPredicate(
            HashMap<Long, Vote> votes,
            Vote vote) {

        HashSet<Long> set = new HashSet<Long>();
        //遍歷已經收到的投票集合，將等於當前投票的集合取出放到set中
        for (Map.Entry<Long,Vote> entry : votes.entrySet()) {
            if (self.getQuorumVerifier().getVotingMembers().containsKey(entry.getKey())
                    && vote.equals(entry.getValue())){
                set.add(entry.getKey());
            }
        }
        
        //統計set，也就是投某個id的票數是否超過一半
        return self.getQuorumVerifier().containsQuorum(set);
    }

    public boolean containsQuorum(Set<Long> ackSet) {
        return (ackSet.size() > half);
    }

最後一關：若是選的是本身，則將本身的狀態更新爲LEADING，不然根據type，要麼是FOLLOWING，要麼是OBSERVING。

到這裏選舉就結束了。

這裏介紹的是一個新集羣啓動時候的選舉過程，啓動的時候就是根據zoo.cfg裏的配置，向各個節點廣播投票，通常都是選投本身。而後收到投票後就會進行進行判斷。若是某個節點收到的投票數超過一半，那麼它就是leader了。 

瞭解了這個過程，咱們來看看另一個問題：

一個集羣有3臺機器，掛了一臺後的影響是什麼？掛了兩臺呢？ 

掛了一臺：掛了一臺後就是收不到其中一臺的投票，可是有兩臺能夠參與投票，按照上面的邏輯，它們開始都投給本身，後來按照選舉的原則，兩我的都投票給其中一個，那麼就有一個節點得到的票等於2，2 > (3/2)=1 的，超過了半數，這個時候是能選出leader的。

掛了兩臺： 掛了兩臺後，怎麼弄也只能得到一張票， 1 不大於 (3/2)=1的，這樣就沒法選出一個leader了。

 

在前面介紹時，爲了簡單我假設的是這是一個嶄新的剛啓動的集羣，這樣的集羣與工做一段時間後的集羣有什麼不一樣呢？不一樣的就是epoch和zxid這兩個參數。在新啓動的集羣裏這兩個通常是相等的，而工做一段時間後這兩個參數有可能有的節點落後其餘節點，至因而爲何，這個還要在後面的存儲和處理額胡斷請求的文章裏介紹。

* 關於邏輯時鐘，咱們的分佈式大牛Leslie Lamport曾寫過一篇論文：Time, Clocks, and the Ordering of Events in a Distributed System