zookeeper-選舉源碼分析

在 zookeeper 集羣中發生選舉的場景有如下三種：

• 集羣啓動時
• Leader 節點重啓時
• Follower 節點重啓時

本文主要針對集羣啓動時發生的選舉實現進行分析。

ZK 集羣中節點在啓動時會調用 QuorumPeer.start方法

public synchronized void start() {
    /**
     * 加載數據文件，獲取 lastProcessedZxid, currentEpoch，acceptedEpoch
     */
    loadDataBase();

    /**
     * 啓動主線程 用於處理客戶端鏈接請求
     */
    cnxnFactory.start();

    /**
     * 開始 leader 選舉; 會相繼建立選舉算法的實現，建立當前節點與集羣中其餘節點選舉通訊的網絡IO，並啓動相應工做線程
     */
    startLeaderElection();

    /**
     * 啓動 QuorumPeer 線程，監聽當前節點服務狀態
     */
    super.start();
}

加載數據文件

在 loadDataBase 方法中，ZK 會經過加載數據文件獲取 lastProcessedZxid , 並經過讀取 currentEpoch , acceptedEpoch 文件來獲取相對應的值；若上述兩文件不存在，則以 lastProcessedZxid 的高 32 位做爲 currentEpoch , acceptedEpoch 值並寫入對應文件中。

初始選舉環境

synchronized public void startLeaderElection() {
    try {
        // 建立投票
        currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
    } catch(IOException e) {
    }
    // 從集羣中節點列表，查找當前節點與其餘進行信息同步的地址
    for (QuorumServer p : getView().values()) {
        if (p.id == myid) {
            myQuorumAddr = p.addr;
            break;
        }
    }
    if (myQuorumAddr == null) {
        throw new RuntimeException("My id " + myid + " not in the peer list");
    }
    
    // electionType == 3
    this.electionAlg = createElectionAlgorithm(electionType);
}

protected Election createElectionAlgorithm(int electionAlgorithm){
    Election le=null;
            
    //TODO: use a factory rather than a switch
    switch (electionAlgorithm) {
        // 忽略其餘算法的實現
    case 3:
        /**
         * 建立 QuorumCnxManager 實例，並啓動 QuorumCnxManager.Listener 線程用於與集羣中其餘節點進行選舉通訊;
         */
        qcm = createCnxnManager();
        QuorumCnxManager.Listener listener = qcm.listener;
        if(listener != null){
            listener.start();
            /**
             * 建立選舉算法 FastLeaderElection 實例
             */
            le = new FastLeaderElection(this, qcm);
        } else {
            LOG.error("Null listener when initializing cnx manager");
        }
        break;
    default:
        assert false;
    }
    return le;
}

初始節點的相關實例以後，執行 super.start() 方法，因 QuorumPeer 類繼承 ZooKeeperThread 故會啓動 QuorumPeer 線程

public void run() {
        // 代碼省略
        try {
            /*
             * Main loop
             */
            while (running) {
                switch (getPeerState()) {
                case LOOKING:
                    LOG.info("LOOKING");

                    if (Boolean.getBoolean("readonlymode.enabled")) {
                        // 只讀模式下代碼省略
                    } else {
                        try {
                            setBCVote(null);
                            setCurrentVote(makeLEStrategy().lookForLeader());
                        } catch (Exception e) {
                            LOG.warn("Unexpected exception", e);
                            setPeerState(ServerState.LOOKING);
                        }
                    }
                    break;
                // 忽略其餘狀態下的處理邏輯
                }
            }
        } finally {
            
        }
    }

選舉

從上述代碼能夠看出 QuorumPeer 線程在運行過程當中輪詢監聽當前節點的狀態並進行相應的邏輯處理，集羣啓動時節點狀態爲 LOOKING (也就是選舉 Leader 過程)，此時會調用 FastLeaderElection.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 邏輯時鐘加一
                logicalclock.incrementAndGet();
                /**
                 * 更新提案信息，用於後續投票；集羣啓動節點默認選舉自身爲 Leader
                 */
                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
            }

            /**
             * 發送選舉投票提案
             */
            sendNotifications();

            /*
             * Loop in which we exchange notifications until we find a leader
             */

            while ((self.getPeerState() == ServerState.LOOKING) &&
                    (!stop)){
                /*
                 * Remove next notification from queue, times out after 2 times
                 * the termination time
                 */
                /**
                 * 從 recvqueue 隊列中獲取外部節點的選舉投票信息
                 */
                Notification n = recvqueue.poll(notTimeout,
                        TimeUnit.MILLISECONDS);

                /*
                 * Sends more notifications if haven't received enough.
                 * Otherwise processes new notification.
                 */
                if(n == null){
                    /**
                     * 檢查上一次發送的選舉投票信息是否所有發送；
                     * 若已發送則從新在發送一遍，反之說明當前節點與集羣中其餘節點未鏈接，則執行 connectAll() 創建鏈接 
                     */
                    if(manager.haveDelivered()){
                        sendNotifications();
                    } else {
                        manager.connectAll();
                    }

                    /*
                     * Exponential backoff
                     */
                    int tmpTimeOut = notTimeout*2;
                    notTimeout = (tmpTimeOut < maxNotificationInterval?
                            tmpTimeOut : maxNotificationInterval);
                    LOG.info("Notification time out: " + notTimeout);
                }
                else if(self.getVotingView().containsKey(n.sid)) {
                    /**
                     * 只處理同一集羣中節點的投票請求
                     */ 
                    switch (n.state) {
                    case LOOKING:
                        // If notification > current, replace and send messages out
                        if (n.electionEpoch > logicalclock.get()) {
                            /**
                             * 外部投票選舉週期大於當前節點選舉週期
                             * 
                             * step1 : 更新選舉週期值
                             * step2 : 清空已收到的選舉投票數據
                             * step3 : 選舉投票 PK，選舉規則參見 totalOrderPredicate 方法
                             * step4 : 變動選舉投票併發送
                             */
                            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)) {
                            /**
                             * 同一選舉週期
                             *                            
                             * step1 : 選舉投票 PK，選舉規則參見 totalOrderPredicate 方法
                             * step2 : 變動選舉投票併發送
                             */
                            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))) {

                            // ......
                            
                            if (n == null) {
                                /**
                                 * 選舉結束判斷當前節點狀態; 若提案的 leader == myid 則 state = LEADING, 反之爲 FOLLOWING 
                                 */
                                self.setPeerState((proposedLeader == self.getId()) ?
                                        ServerState.LEADING: learningState());
                                // 變動當前投票信息
                                Vote endVote = new Vote(proposedLeader,
                                                        proposedZxid,
                                                        logicalclock.get(),
                                                        proposedEpoch);
                                leaveInstance(endVote);
                                return endVote;
                            }
                        }
                        break;
                    case OBSERVING:
                        LOG.debug("Notification from observer: " + n.sid);
                        break;
                    case FOLLOWING:
                    case LEADING:
                        // ...... 
                        break;
                    default:
                        LOG.warn("Notification state unrecognized: {} (n.state), {} (n.sid)",
                                n.state, n.sid);
                        break;
                    }
                } else {
                    LOG.warn("Ignoring notification from non-cluster member " + n.sid);
                }
            }
            return null;
        } finally {
            // ......
        }
    }

從 lookForLeader 方法的實現能夠看出，選舉流程以下：

• 發送內部投票

  內部投票發送邏輯參考後續小節

• 接收外部投票

  接收外部投票邏輯參考後續小節

• 選舉投票 PK

  當接收到外部節點投票信息後會與內部投票信息進行 PK 已肯定投票優先權；PK 規則參見 totalOrderPredicate 方法以下

protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
    if(self.getQuorumVerifier().getWeight(newId) == 0){
        return false;
    }
    
    /*
     * We return true if one of the following three cases hold:
     * 1- New epoch is higher
     * 2- New epoch is the same as current epoch, but new zxid is higher
     * 3- New epoch is the same as current epoch, new zxid is the same
     *  as current zxid, but server id is higher.
     */
    return ((newEpoch > curEpoch) || 
            ((newEpoch == curEpoch) &&
            ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))));
}

從其實現能夠看出選舉投票 PK 規則以下：

* 比較外部投票與內部投票的選舉週期值，選舉週期大的值優先
* 若選舉週期值一致，則比較事務 ID； 事務 ID 最新的優先
* 若選舉週期值一致且事務 ID 值相同，則比較投票節點的 server id; server id 最大的優先

• 統計選舉投票

  當接收到外部投票以後，都會統計下此輪選舉的投票狀況並判斷是否可結束選舉; 參考 termPredicate 方法

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

    HashSet<Long> set = new HashSet<Long>();

    /**
     * 統計接收的投票中與當前節點所推舉 leader 投票一致的個數
     */
    for (Map.Entry<Long,Vote> entry : votes.entrySet()) {
        if (vote.equals(entry.getValue())){
            set.add(entry.getKey());
        }
    }

    /**
     * 若是超過一半的投票一致 則說明能夠終止本次選舉
     */
    return self.getQuorumVerifier().containsQuorum(set);
}

• 確認節點角色

  當此輪選舉結束以後，經過判斷所推舉的 leader server id 是否與當前節點 server id 相等； 若相等則說明當前節點爲 leader, 反之爲 follower。

發送接收投票

上文中主要聊了下 ZK 選舉算法的核心部分，下面接着看下集羣節點在選舉過程當中是如何發送本身的投票和接收外部的投票及相關處理邏輯。

首先經過 FastLeaderElection.sendNotifications 方法看下發送投票邏輯：

private void sendNotifications() {
    for (QuorumServer server : self.getVotingView().values()) {
        long sid = server.id;

        /**
         * 發送投票通知信息
         *
         * leader : 被推舉的服務器 myid
         * zxid : 被推舉的服務器 zxid
         * electionEpoch : 當前節點選舉週期
         * ServerState state : 當前節點狀態
         * sid : 消息接收方 myid
         * peerEpoch : 被推舉的服務器 epoch
         */
        ToSend notmsg = new ToSend(ToSend.mType.notification,
                proposedLeader,
                proposedZxid,
                logicalclock.get(),
                QuorumPeer.ServerState.LOOKING,
                sid,
                proposedEpoch);

        /**
         * 將消息添加到隊列 sendqueue 中;
         *
         * @see Messenger.WorkerSender sendqueue 隊列會被 WorkerSender 消費
         */
        sendqueue.offer(notmsg);
    }
}

從實現能夠看出節點在啓動階段會將自身信息封裝爲 ToSend 實例（也就是選舉自身爲 leader）並添加到隊列 FastLeaderElection.sendqueue 中；那麼此時咱們會問到 FastLeaderElection.sendqueue 隊列中的消息被誰消費處理呢 ？ 讓咱們回過頭看下節點在啓動初始化選舉環境時建立 QuorumCnxManager, FastLeaderElection 實例的過程。

PS : FastLeaderElection.sendqueue 隊列中消息被誰消費 ？

QuorumCnxManager

public QuorumCnxManager(final long mySid,
                            Map<Long,QuorumPeer.QuorumServer> view,
                            QuorumAuthServer authServer,
                            QuorumAuthLearner authLearner,
                            int socketTimeout,
                            boolean listenOnAllIPs,
                            int quorumCnxnThreadsSize,
                            boolean quorumSaslAuthEnabled,
                            ConcurrentHashMap<Long, SendWorker> senderWorkerMap) {
    this.senderWorkerMap = senderWorkerMap;
    this.recvQueue = new ArrayBlockingQueue<Message>(RECV_CAPACITY);
    this.queueSendMap = new ConcurrentHashMap<Long, ArrayBlockingQueue<ByteBuffer>>();

    this.lastMessageSent = new ConcurrentHashMap<Long, ByteBuffer>();
    String cnxToValue = System.getProperty("zookeeper.cnxTimeout");
    if(cnxToValue != null){
        this.cnxTO = Integer.parseInt(cnxToValue);
    }

    this.mySid = mySid;
    this.socketTimeout = socketTimeout;
    this.view = view;
    this.listenOnAllIPs = listenOnAllIPs;

    initializeAuth(mySid, authServer, authLearner, quorumCnxnThreadsSize,
            quorumSaslAuthEnabled);

    listener = new Listener();
}

在 QuorumCnxManager 實例化後，會啓動一個 QuorumCnxManager.Listener 線程；同時在 QuorumCnxManager 實例中存在三個重要的集合容器變量：

• senderWorkerMap : 發送器集合，Map 類型按 server id 分組；爲集羣中的每一個節點分配一個 SendWorker 負責消息的發送
• recvQueue ： 消息接收隊列，用於存放從外部節點接收到的投票消息
• queueSendMap ： 消息發送隊列，Map 類型按 server id 分組；爲集羣中的每一個節點分配一個阻塞隊列存放待發送的消息，從而保證各個節點之間的消息發送互不影響

下面咱們再看下 QuorumCnxManager.Listener 線程啓動後，主要作了什麼:

public void run() {
    int numRetries = 0;
    InetSocketAddress addr;
    while((!shutdown) && (numRetries < 3)){
        try {
            ss = new ServerSocket();
            ss.setReuseAddress(true);

            /**
             * 獲取當前節點的選舉地址並 bind 監聽等待外部節點鏈接
             */
            addr = view.get(QuorumCnxManager.this.mySid).electionAddr;
            ss.bind(addr);

            while (!shutdown) {

                /**
                 * 接收外部節點鏈接並處理
                 */
                Socket client = ss.accept();
                setSockOpts(client);                
                receiveConnection(client);

                numRetries = 0;
            }
        } catch (IOException e) {
            LOG.error("Exception while listening", e);
            numRetries++;
            ss.close();
            Thread.sleep(1000);
        }
    }
}

跟蹤代碼發現 receiveConnection 方法最終會調用方法 handleConnection 以下

private void handleConnection(Socket sock, DataInputStream din)
            throws IOException {
    /**
     * 讀取外部節點的 server id 
     * ps : 此時的 server id 是何時發送的呢 ？
     */
    Long sid = din.readLong();
  
    if (sid < this.mySid) {
        /**
         * 若外部節點的 server id 小於當前節點的 server id，則關閉此鏈接，改成由當前節點發起鏈接
         * ps ： 該限制說明選舉過程當中，zk 只容許 server id 較大的一方去主動發起鏈接避免重複鏈接
         */
        SendWorker sw = senderWorkerMap.get(sid);
        if (sw != null) {
            sw.finish();
        }

        closeSocket(sock);
        connectOne(sid);
    } else {
        SendWorker sw = new SendWorker(sock, sid);
        RecvWorker rw = new RecvWorker(sock, din, sid, sw);
        sw.setRecv(rw);

        SendWorker vsw = senderWorkerMap.get(sid);
        
        if(vsw != null)
            vsw.finish();
        
        /**
         * 按 server id 分組，爲外部節點分配 SendWorker, RecvWorker 和一個消息發送隊列
         */
        senderWorkerMap.put(sid, sw);
        queueSendMap.putIfAbsent(sid, new ArrayBlockingQueue<ByteBuffer>(SEND_CAPACITY));
        
        /**
         * 啓動外部節點對應的 SendWorker， RecvWorker 線程
         */
        sw.start();
        rw.start();
        
        return;
    }
}

至此會發現 QuorumCnxManager.Listener 線程處理邏輯以下：

• 監聽當前節點的 election address 等待接收外部節點鏈接
• 讀取外部節點的 server id 並與當前節點的 server id 比較；若前者小則關閉鏈接，改由當前節點發起鏈接
• 反之爲外部節點分配 SendWorker，RecvWorker 線程及消息發送隊列

PS ： 此處咱們會有個疑問外部節點的 server id 是何時發送過來的呢 ？

下面咱們在看下爲每一個外部節點開啓了 SendWorker， RecvWorker 線程後作了什麼：

• SendWorker

public void run() {
    // 省略
    try {
        while (running && !shutdown && sock != null) {

            ByteBuffer b = null;
            try {
                /**
                 * 經過 server id 獲取待發送給集羣中節點的消息隊列
                 */
                ArrayBlockingQueue<ByteBuffer> bq = queueSendMap
                        .get(sid);
                if (bq != null) {
                    /**
                     * 從隊列中獲取待發送的消息
                     */
                    b = pollSendQueue(bq, 1000, TimeUnit.MILLISECONDS);
                } else {
                    LOG.error("No queue of incoming messages for " +
                              "server " + sid);
                    break;
                }

                if(b != null){
                    lastMessageSent.put(sid, b);
                    /**
                     * 寫入 socket 的輸出流完成消息的發送
                     */
                    send(b);
                }
            } catch (InterruptedException e) {               
            }
        }
    } catch (Exception e) {        
    }
}

synchronized void send(ByteBuffer b) throws IOException {
    byte[] msgBytes = new byte[b.capacity()];
    try {
        b.position(0);
        b.get(msgBytes);
    } catch (BufferUnderflowException be) {
        LOG.error("BufferUnderflowException ", be);
        return;
    }
    /**
     * 發送的報文包括：消息體正文長度和消息體正文
     */
    dout.writeInt(b.capacity());
    dout.write(b.array());
    dout.flush();
}

經過代碼實現咱們知道 SendWorker 的職責就是從 queueSendMap 隊列中獲取待發送給遠程節點的消息並執行發送。

PS : 此處咱們會有個疑問 QuorumCnxManager.queueSendMap 中節點對應隊列中待發送的消息是誰生產的呢 ？

• RecvWorker

public void run() {
    threadCnt.incrementAndGet();
    try {
        while (running && !shutdown && sock != null) {
            /**
             * 讀取外部節點發送的消息
             * 由 SendWorker 可知前 4 字節爲消息載體有效長度
             */
            int length = din.readInt();
            if (length <= 0 || length > PACKETMAXSIZE) {
                throw new IOException(
                        "Received packet with invalid packet: "
                                + length);
            }
            /**
             * 讀取消息體正文
             */
            byte[] msgArray = new byte[length];
            din.readFully(msgArray, 0, length);
            ByteBuffer message = ByteBuffer.wrap(msgArray);
            /**
             * 將讀取的消息包裝爲 Message 對象添加到隊列 recvQueue 中
             */
            addToRecvQueue(new Message(message.duplicate(), sid));
        }
    } catch (Exception e) {
        LOG.warn("Connection broken for id " + sid + ", my id = "
                 + QuorumCnxManager.this.mySid + ", error = " , e);
    } finally {
        LOG.warn("Interrupting SendWorker");
        sw.finish();
        if (sock != null) {
            closeSocket(sock);
        }
    }
}

public void addToRecvQueue(Message msg) {
    synchronized(recvQLock) {
        // 省略
        try {
            recvQueue.add(msg);
        } catch (IllegalStateException ie) {
            // This should never happen
            LOG.error("Unable to insert element in the recvQueue " + ie);
        }
    }
}

從上面能夠看出 RecvWorker 線程在運行期間會接收 server id 對應的外部節點發送的消息，並將其放入 QuorumCnxManager.recvQueue 隊列中。
到目前爲止咱們基本完成對 QuorumCnxManager 核心功能的分析，發現其功能主要是負責集羣中當前節點與外部節點進行選舉通信的網絡 IO 操做，譬如接收外部節點選舉投票和向外部節點發送內部投票。

FastLeaderElection

下面咱們在接着回頭看下 FastLeaderElection 類實例的過程：

public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager){
    this.stop = false;
    this.manager = manager;
    starter(self, manager);
}

private void starter(QuorumPeer self, QuorumCnxManager manager) {
    this.self = self;
    proposedLeader = -1;
    proposedZxid = -1;

    sendqueue = new LinkedBlockingQueue<ToSend>();
    recvqueue = new LinkedBlockingQueue<Notification>();
    this.messenger = new Messenger(manager);
}

Messenger(QuorumCnxManager manager) {
    /**
     * 啓動 WorkerSender 線程用於發送消息
     */
    this.ws = new WorkerSender(manager);

    Thread t = new Thread(this.ws,
            "WorkerSender[myid=" + self.getId() + "]");
    t.setDaemon(true);
    t.start();

    /**
     * 啓動 WorkerReceiver 線程用於接收消息
     */
    this.wr = new WorkerReceiver(manager);

    t = new Thread(this.wr,
            "WorkerReceiver[myid=" + self.getId() + "]");
    t.setDaemon(true);
    t.start();
}

從 FastLeaderElection 實例化過程咱們知道，其內部分別啓動了線程 WorkerSender，WorkerReceiver ；那麼接下來看下這兩個線程具體作什麼吧。

WorkerSender

public void run() {
    while (!stop) {
        try {
            /**
             * 從 sendqueue 隊列中獲取 ToSend 待發送的消息
             */ 
            ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
            if(m == null) continue;

            process(m);
        } catch (InterruptedException e) {
            break;
        }
    }
    LOG.info("WorkerSender is down");
}

void process(ToSend m) {
    // 將 ToSend 轉換爲 40字節 ByteBuffer
    ByteBuffer requestBuffer = buildMsg(m.state.ordinal(), 
                                            m.leader,
                                            m.zxid, 
                                            m.electionEpoch, 
                                            m.peerEpoch);
    // 交由 QuorumCnxManager 執行發送
    manager.toSend(m.sid, requestBuffer);
}

看了 WorkerSender 的實現是否是明白了什麼？ 還記得上文中 FastLeaderElection.sendNotifications 方法執行發送通知的時候的疑惑嗎 ？ FastLeaderElection.sendqueue 隊列產生的消息就是被 WorkerSender 線程所消費處理, WorkerSender 會將消息轉發至 QuorumCnxManager 處理

public void toSend(Long sid, ByteBuffer b) {
    /*
     * If sending message to myself, then simply enqueue it (loopback).
     * 若是是發給本身的投票，則將其添加到接收隊列中等待處理
     */
    if (this.mySid == sid) {
         b.position(0);
         addToRecvQueue(new Message(b.duplicate(), sid));
        /*
         * Otherwise send to the corresponding thread to send.
         */
    } else {
         /*
          * Start a new connection if doesn't have one already.
          */
         ArrayBlockingQueue<ByteBuffer> bq = new ArrayBlockingQueue<ByteBuffer>(SEND_CAPACITY);
         ArrayBlockingQueue<ByteBuffer> bqExisting = queueSendMap.putIfAbsent(sid, bq);

         // 將發送的消息放入對應的隊列中，若隊列滿了則將隊列頭部元素移除
         if (bqExisting != null) {
             addToSendQueue(bqExisting, b);
         } else {
             addToSendQueue(bq, b);
         }
         connectOne(sid);
            
    }
}

private void addToSendQueue(ArrayBlockingQueue<ByteBuffer> queue,
          ByteBuffer buffer) {
    // 省略
    try {
        // 將消息插入節點對應的隊列中
        queue.add(buffer);
    } catch (IllegalStateException ie) {
    }
}

QuorumCnxManager 在收到 FastLeaderElection.WorkerSender 轉發的消息時，會判斷當前消息是否發給本身的投票，如果則將消息添加到接收隊列中，反之會將消息添加到 queueSendMap 對應 server id 的隊列中；看到這裏的時候是否是就明白了在 QuorumCnxManager.SendWorker 分析時候的疑惑呢 。 這個時候投票消息未必可以發送出去，由於當前節點與外部節點的通道是否已創建還未知，因此繼續執行 connectOne

synchronized public void connectOne(long sid){
    /**
     * 判斷當前服務節點是否與 sid 外部服務節點創建鏈接;有可能對方先發起鏈接
     * 若已鏈接則等待後續處理，反之發起鏈接
     */
    if (!connectedToPeer(sid)){
        InetSocketAddress electionAddr;
        if (view.containsKey(sid)) {
            electionAddr = view.get(sid).electionAddr;
        } else {
            LOG.warn("Invalid server id: " + sid);
            return;
        }
        try {

            LOG.debug("Opening channel to server " + sid);
            Socket sock = new Socket();
            setSockOpts(sock);
            sock.connect(view.get(sid).electionAddr, cnxTO);
            LOG.debug("Connected to server " + sid);

            initiateConnection(sock, sid);

        } catch (UnresolvedAddressException e) {
           
        } catch (IOException e) {
           
        }
    } else {
        LOG.debug("There is a connection already for server " + sid);
    }
}

public boolean connectedToPeer(long peerSid) {
    return senderWorkerMap.get(peerSid) != null;
}

private boolean startConnection(Socket sock, Long sid)
            throws IOException {
    DataOutputStream dout = null;
    DataInputStream din = null;
    try {
        /**
         * 發送當前節點的 server id，需告知對方我是哪臺節點
         */
        dout = new DataOutputStream(sock.getOutputStream());
        dout.writeLong(this.mySid);
        dout.flush();

        din = new DataInputStream(
                new BufferedInputStream(sock.getInputStream()));
    } catch (IOException e) {
        LOG.warn("Ignoring exception reading or writing challenge: ", e);
        closeSocket(sock);
        return false;
    }

    // 只容許 sid 值大的服務器去主動和其餘服務器鏈接，不然斷開鏈接
    if (sid > this.mySid) {
        LOG.info("Have smaller server identifier, so dropping the " +
                 "connection: (" + sid + ", " + this.mySid + ")");
        closeSocket(sock);
        // Otherwise proceed with the connection
    } else {
        SendWorker sw = new SendWorker(sock, sid);
        RecvWorker rw = new RecvWorker(sock, din, sid, sw);
        sw.setRecv(rw);

        SendWorker vsw = senderWorkerMap.get(sid);
        
        if(vsw != null)
            vsw.finish();
        
        senderWorkerMap.put(sid, sw);
        queueSendMap.putIfAbsent(sid, new ArrayBlockingQueue<ByteBuffer>(SEND_CAPACITY));
        
        sw.start();
        rw.start();
        
        return true;    
        
    }
    return false;
}

從上述代碼能夠看出節點在與外部節點鏈接後會先發送 myid 報文告知對方我是哪一個節點（這也是爲何 QuorumCnxManager.Listener 線程在接收到一個鏈接請求時會先執行 getLong 獲取 server id 了）；一樣在鏈接創建的時候也遵循一個原則（只容許 server id 較大的一方發起鏈接）。

WorkerReceiver

public void run() {

    Message response;
    while (!stop) {
        // Sleeps on receive
        try{
            /**
             * 從 QuorumCnxManager.recvQueue 隊列中獲取接收的外部投票
             */
            response = manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
            if(response == null) continue;
          
            if(!self.getVotingView().containsKey(response.sid)){
                // 忽略對方是觀察者的處理
            } else {
                // Instantiate Notification and set its attributes
                Notification n = new Notification();
                
                   // 將 message 轉成 notification 對象

                if(self.getPeerState() == QuorumPeer.ServerState.LOOKING){
                    // 當前節點狀態爲 looking，則將外部節點投票添加到 recvqueue 隊列中
                    recvqueue.offer(n);

                    if((ackstate == QuorumPeer.ServerState.LOOKING)
                            && (n.electionEpoch < logicalclock.get())){
                        // 若外部節點選舉週期小於當前節點選舉週期則發送內部投票
                        Vote v = getVote();
                        ToSend notmsg = new ToSend(ToSend.mType.notification,
                                v.getId(),
                                v.getZxid(),
                                logicalclock.get(),
                                self.getPeerState(),
                                response.sid,
                                v.getPeerEpoch());
                        sendqueue.offer(notmsg);
                    }
                } else {
                    // 忽略其餘狀態時的處理
                }
            }
        } catch (InterruptedException e) {
        }
    }
    LOG.info("WorkerReceiver is down");
}

此時咱們明白 WorkerReceiver 線程在運行期間會一直從 QuorumCnxManager.recvQueue 的隊列中拉取接收到的外部投票信息，若當前節點爲 LOOKING 狀態，則將外部投票信息添加到 FastLeaderElection.recvqueue 隊列中，等待 FastLeaderElection.lookForLeader 選舉算法處理投票信息。

到此咱們基本明白了 ZK 集羣節點發送和接收投票的處理流程，可是這個時候您是否是又有一種懵的狀態呢 笑哭，咱們會發現選舉過程當中依賴了多個線程 WorkerSender, SendWorker, WorkerReceiver, RecvWorker ，多個阻塞隊列 sendqueue, recvqueue, queueSendMap, recvQueue 並且名字起的很相似，更讓人懵 ； 不過莫慌，咱們來經過下面的圖來縷下思路

小結

看了這麼長時間的代碼，也夠累的；最後咱們就來個小結吧 ：

• QuorumCnxManager 類主要職能是負責集羣中節點與外部節點進行通訊及投票信息的中轉
• FastLeaderElection 類是選舉投票的核心實現

• 選舉投票規則

  • 比較外部投票與內部投票的選舉週期值，選舉週期大的值優先
  • 若選舉週期值一致，則比較事務 ID； 事務 ID 最新的優先
  • 若選舉週期值一致且事務 ID 值相同，則比較投票節點的 server id; server id 最大的優先
• 集羣中節點通訊時爲了不重複創建鏈接，遵照一個原則：鏈接老是由 server id 較大的一方發起