zookeeper 的 Leader 選舉源碼解析

導讀

zookeeper 是一個分佈式服務框架，主要解決分佈式應用中常見的多種數據問題，例如集羣管理，狀態同步等。爲解決這些問題 zookeeper 需要 Leader 選舉進行保障數據的強一致性機制和穩定性。本文通過集羣的配置，對 leader 選舉源進行解析，讓讀者們瞭解如何利用 BIO 通信機制，多線程多層隊列實現高性能架構。

01 Leader 選舉機制

在今年的敏捷團隊建設中，我通過 Suite 執行器實現了一鍵自動化單元測試。Juint 除了 Suite 執行器還有哪些執行器呢？由此我的 Runner 探索之旅開始了！

Leader 選舉機制採用半數選舉算法。

每一個 zookeeper 服務端稱之爲一個節點，每個節點都有投票權，把其選票投向每一個有選舉權的節點，當其中一個節點選舉出票數過半，這個節點就會成爲 Leader，其它節點成爲 Follower。

02 Leader 選舉集羣配置

理解，首先 MCube 會依據模板緩存狀態判斷是否需要網絡獲取最新模板，當獲取到模板後進行模板加載，加載階段會將產物轉換爲視圖樹的結構，轉換完成後將通過表達式引擎解析表達式並取得正確的值，通過事件解析引擎解析用戶自定義事件並完成事件的綁定，完成解析賦值以及事件綁定後進行視圖的渲染，最終將目標頁面展示到屏幕。

重命名 zoo_sample.cfg 文件爲 zoo1.cfg ，zoo2.cfg，zoo3.cfg，zoo4.cfg
修改 zoo.cfg 文件，修改值如下：

【plain】
zoo1.cfg文件內容：
dataDir=/export/data/zookeeper-1
clientPort=2181
server.1=127.0.0.1:2001:3001
server.2=127.0.0.1:2002:3002:participant
server.3=127.0.0.1:2003:3003:participant
server.4=127.0.0.1:2004:3004:observer
zoo2.cfg文件內容：
dataDir=/export/data/zookeeper-2
clientPort=2182
server.1=127.0.0.1:2001:3001
server.2=127.0.0.1:2002:3002:participant
server.3=127.0.0.1:2003:3003:participant
server.4=127.0.0.1:2004:3004:observer
zoo3.cfg文件內容：
dataDir=/export/data/zookeeper-3
clientPort=2183
server.1=127.0.0.1:2001:3001
server.2=127.0.0.1:2002:3002:participant
server.3=127.0.0.1:2003:3003:participant
server.4=127.0.0.1:2004:3004:observer
zoo4.cfg文件內容：
dataDir=/export/data/zookeeper-4
clientPort=2184
server.1=127.0.0.1:2001:3001
server.2=127.0.0.1:2002:3002:participant
server.3=127.0.0.1:2003:3003:participant
server.4=127.0.0.1:2004:3004:observer

server. 第幾號服務器 (對應 myid 文件內容)=ip: 數據同步端口: 選舉端口: 選舉標識

participant 默認參與選舉標識，可不寫. observer 不參與選舉

在 /export/data/zookeeper-1,/export/data/zookeeper-2,/export/data/zookeeper-3,/export/data/zookeeper-4 目錄下創建 myid 文件，文件內容分別寫 1 ，2，3，4，用於標識 sid(全稱：Server ID) 賦值。
啓動三個 zookeeper 實例：

bin/zkServer.sh start conf/zoo1.cfg
bin/zkServer.sh start conf/zoo2.cfg
bin/zkServer.sh start conf/zoo3.cfg

每啓動一個實例，都會讀取啓動參數配置 zoo.cfg 文件，這樣實例就可以知道其作爲服務端身份信息 sid 以及集羣中有多少個實例參與選舉。

03 Leader 選舉流程

圖 1 第一輪到第二輪投票流程

前提：

設定票據數據格式 vote(sid,zxid,epoch)

sid 是 Server ID 每臺服務的唯一標識，是 myid 文件內容；
zxid 是數據事務 id 號；
epoch 爲選舉週期，爲方便理解下面講解內容暫定爲 1 初次選舉，不寫入下面內容裏。

按照順序啓動 sid=1，sid=2 節點

第一輪投票：

sid=1 節點：初始選票爲自己，將選票 vote(1,0) 發送給 sid=2 節點；
sid=2 節點：初始選票爲自己，將選票 vote(2,0) 發送給 sid=1 節點；
sid=1 節點：收到 sid=2 節點選票 vote(2,0) 和當前自己的選票 vote(1,0)，首先比對 zxid 值，zxid 越大代表數據最新，優先選擇 zxid 最大的選票，如果 zxid 相同，選舉最大 sid。當前投票選舉結果爲 vote(2,0)，sid=1 節點的選票變爲 vote(2,0)；
sid=2 節點：收到 sid=1 節點選票 vote(1,0) 和當前自己的選票 vote(2,0), 參照上述選舉方式，選舉結果爲 vote(2,0),sid=2 節點的選票不變；
第一輪投票選舉結束。

第二輪投票：

sid=1 節點：當前自己的選票爲 vote(2,0), 將選票 vote(2,0) 發送給 sid=2 節點；
sid=2 節點：當前自己的選票爲 vote(2,0), 將選票 vote(2,0) 發送給 sid=1 節點；
sid=1 節點：收到 sid=2 節點選票 vote(2,0) 和自己的選票 vote(2,0), 按照半數選舉算法，總共 3 個節點參與選舉，已有 2 個節點選舉出相同選票, 推舉 sid=2 節點爲 Leader，自己角色變爲 Follower；
sid=2 節點：收到 sid=1 節點選票 vote(2,0) 和自己的選票 vote(2,0)，按照半數選舉算法推舉 sid=2 節點爲 Leader, 自己角色變爲 Leader。

這時啓動 sid=3 節點後，集羣裏已經選舉出 leader，sid=1 和 sid=2 節點會將自己的 leader 選票發回給 sid=3 節點，通過半數選舉結果還是 sid=2 節點爲 leader。

3.1 Leader 選舉採用多層隊列架構

zookeeper 選舉底層主要分爲選舉應用層和消息傳輸隊列層，第一層應用層隊列統一接收和發送選票，而第二層傳輸層隊列，是按照服務端 sid 分成了多個隊列，是爲了避免給每臺服務端發送消息互相影響。比如對某臺機器發送不成功不會影響正常服務端的發送。

圖 2 多層隊列上下關係交互流程圖

04 解析代碼入口類

通過查看 zkServer.sh 文件內容找到服務啓動類：

org.apache.zookeeper.server.quorum.QuorumPeerMain

05 選舉流程代碼解析

圖 3 選舉代碼實現流程圖

加載配置文件 QuorumPeerConfig.parse(path)；

針對 Leader 選舉關鍵配置信息如下：

讀取 dataDir 目錄找到 myid 文件內容，設置當前應用 sid 標識，做爲投票人身份信息。下面遇到 myid 變量爲當前節點自己 sid 標識。
設置 peerType 當前應用是否參與選舉
new QuorumMaj() 解析 server. 前綴加載集羣成員信息，加載 allMembers 所有成員，votingMembers 參與選舉成員，observingMembers 觀察者成員，設置 half 值 votingMembers.size()/2.

【Java】
public QuorumMaj(Properties props) throws ConfigException {
        for (Entry<Object, Object> entry : props.entrySet()) {
            String key = entry.getKey().toString();
            String value = entry.getValue().toString();
            //讀取集羣配置文件中的server.開頭的應用實例配置信息
            if (key.startsWith("server.")) {
                int dot = key.indexOf('.');
                long sid = Long.parseLong(key.substring(dot + 1));
                QuorumServer qs = new QuorumServer(sid, value);
                allMembers.put(Long.valueOf(sid), qs);
                if (qs.type == LearnerType.PARTICIPANT)
//應用實例綁定的角色爲PARTICIPANT意爲參與選舉
                    votingMembers.put(Long.valueOf(sid), qs);
                else {
                    //觀察者成員
                    observingMembers.put(Long.valueOf(sid), qs);
                }
            } else if (key.equals("version")) {
                version = Long.parseLong(value, 16);
            }
        }
        //過半基數
        half = votingMembers.size() / 2;
    }

QuorumPeerMain.runFromConfig(config) 啓動服務；
QuorumPeer.startLeaderElection() 開啓選舉服務；

設置當前選票 new Vote(sid,zxid,epoch)

【plain】
synchronized public void startLeaderElection(){
try {
           if (getPeerState() == ServerState.LOOKING) {
               //首輪：當前節點默認投票對象爲自己
               currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
           }
       } catch(IOException e) {
           RuntimeException re = new RuntimeException(e.getMessage());
           re.setStackTrace(e.getStackTrace());
           throw re;
       }
//........
}

創建選舉管理類：QuorumCnxnManager；
初始化 recvQueue<Message(sid,ByteBuffer)> 接收投票隊列 (第二層傳輸隊列)；
初始化 queueSendMap<sid,queue> 按 sid 發送投票隊列 (第二層傳輸隊列)；
初始化 senderWorkerMap<sid,SendWorker> 發送投票工作線程容器，表示着與 sid 投票節點已連接；
初始化選舉監聽線程類 QuorumCnxnManager.Listener。

【Java】
//QuorumPeer.createCnxnManager()
public QuorumCnxManager(QuorumPeer self,
                        final long mySid,
                        Map<Long,QuorumPeer.QuorumServer> view,
                        QuorumAuthServer authServer,
                        QuorumAuthLearner authLearner,
                        int socketTimeout,
                        boolean listenOnAllIPs,
                        int quorumCnxnThreadsSize,
                        boolean quorumSaslAuthEnabled) {
    //接收投票隊列(第二層傳輸隊列)
    this.recvQueue = new ArrayBlockingQueue<Message>(RECV_CAPACITY);
    //按sid發送投票隊列(第二層傳輸隊列)
    this.queueSendMap = new ConcurrentHashMap<Long, ArrayBlockingQueue<ByteBuffer>>();
    //發送投票工作線程容器，表示着與sid投票節點已連接 
    this.senderWorkerMap = new ConcurrentHashMap<Long, SendWorker>();
    this.lastMessageSent = new ConcurrentHashMap<Long, ByteBuffer>();
    String cnxToValue = System.getProperty("zookeeper.cnxTimeout");
    if(cnxToValue != null){
        this.cnxTO = Integer.parseInt(cnxToValue);
    }
    this.self = self;
    this.mySid = mySid;
    this.socketTimeout = socketTimeout;
    this.view = view;
    this.listenOnAllIPs = listenOnAllIPs;
    initializeAuth(mySid, authServer, authLearner, quorumCnxnThreadsSize,
            quorumSaslAuthEnabled);
    // Starts listener thread that waits for connection requests 
    //創建選舉監聽線程 接收選舉投票請求
    listener = new Listener();
    listener.setName("QuorumPeerListener");
}
//QuorumPeer.createElectionAlgorithm
protected Election createElectionAlgorithm(int electionAlgorithm){
    Election le=null;
    //TODO: use a factory rather than a switch
    switch (electionAlgorithm) {
    case 0:
        le = new LeaderElection(this);
        break;
    case 1:
        le = new AuthFastLeaderElection(this);
        break;
    case 2:
        le = new AuthFastLeaderElection(this, true);
        break;
    case 3:
        qcm = createCnxnManager();// new QuorumCnxManager(... new Listener())
        QuorumCnxManager.Listener listener = qcm.listener;
        if(listener != null){
            listener.start();//啓動選舉監聽線程
            FastLeaderElection fle = new FastLeaderElection(this, qcm);
            fle.start();
            le = fle;
        } else {
            LOG.error("Null listener when initializing cnx manager");
        }
        break;
    default:
        assert false;
    }
return le;}

開啓選舉監聽線程 QuorumCnxnManager.Listener；

創建 ServerSockket 等待大於自己 sid 節點連接，連接信息存儲到 senderWorkerMap<sid,SendWorker>；
sid>self.sid 纔可以連接過來。

【Java】
//上面的listener.start()執行後，選擇此方法
public void run() {
    int numRetries = 0;
    InetSocketAddress addr;
    Socket client = null;
    while((!shutdown) && (numRetries < 3)){
        try {
            ss = new ServerSocket();
            ss.setReuseAddress(true);
            if (self.getQuorumListenOnAllIPs()) {
                int port = self.getElectionAddress().getPort();
                addr = new InetSocketAddress(port);
            } else {
                // Resolve hostname for this server in case the
                // underlying ip address has changed.
                self.recreateSocketAddresses(self.getId());
                addr = self.getElectionAddress();
            }
            LOG.info("My election bind port: " + addr.toString());
            setName(addr.toString());
            ss.bind(addr);
            while (!shutdown) {
                client = ss.accept();
                setSockOpts(client);
                LOG.info("Received connection request "
                        + client.getRemoteSocketAddress());
                // Receive and handle the connection request
                // asynchronously if the quorum sasl authentication is
                // enabled. This is required because sasl server
                // authentication process may take few seconds to finish,
                // this may delay next peer connection requests.
                if (quorumSaslAuthEnabled) {
                    receiveConnectionAsync(client);
                } else {
//接收連接信息
                    receiveConnection(client);
                }
                numRetries = 0;
            }
        } catch (IOException e) {
            if (shutdown) {
                break;
            }
            LOG.error("Exception while listening", e);
            numRetries++;
            try {
                ss.close();
                Thread.sleep(1000);
            } catch (IOException ie) {
                LOG.error("Error closing server socket", ie);
            } catch (InterruptedException ie) {
                LOG.error("Interrupted while sleeping. " +
                    "Ignoring exception", ie);
            }
            closeSocket(client);
        }
    }
    LOG.info("Leaving listener");
    if (!shutdown) {
        LOG.error("As I'm leaving the listener thread, "
                + "I won't be able to participate in leader "
                + "election any longer: "
                + self.getElectionAddress());
    } else if (ss != null) {
        // Clean up for shutdown.
        try {
            ss.close();
        } catch (IOException ie) {
            // Don't log an error for shutdown.
            LOG.debug("Error closing server socket", ie);
        }
    }
}
//代碼執行路徑：receiveConnection()->handleConnection(...)
private void handleConnection(Socket sock, DataInputStream din)
            throws IOException {
//...省略
     if (sid < self.getId()) {
            /*
             * This replica might still believe that the connection to sid is
             * up, so we have to shut down the workers before trying to open a
             * new connection.
             */
            SendWorker sw = senderWorkerMap.get(sid);
            if (sw != null) {
                sw.finish();
            }
            /*
             * Now we start a new connection
             */
            LOG.debug("Create new connection to server: {}", sid);
            closeSocket(sock);
            if (electionAddr != null) {
                connectOne(sid, electionAddr);
            } else {
                connectOne(sid);
            }
        } else { // Otherwise start worker threads to receive data.
            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();
            }
  //存儲連接信息<sid,SendWorker>
            senderWorkerMap.put(sid, sw);
            queueSendMap.putIfAbsent(sid,
                    new ArrayBlockingQueue<ByteBuffer>(SEND_CAPACITY));
            sw.start();
            rw.start();
     }
}

創建 FastLeaderElection 快速選舉服務；

初始選票發送隊列 sendqueue（第一層隊列）
初始選票接收隊列 recvqueue（第一層隊列）
創建線程 WorkerSender
創建線程 WorkerReceiver

【Java】
//FastLeaderElection.starter
private void starter(QuorumPeer self, QuorumCnxManager manager) {
    this.self = self;
    proposedLeader = -1;
    proposedZxid = -1;
    //發送隊列sendqueue（第一層隊列）
    sendqueue = new LinkedBlockingQueue<ToSend>();
    //接收隊列recvqueue（第一層隊列）
    recvqueue = new LinkedBlockingQueue<Notification>();
    this.messenger = new Messenger(manager);
}
//new Messenger(manager)
Messenger(QuorumCnxManager manager) {
    //創建線程WorkerSender
    this.ws = new WorkerSender(manager);
    this.wsThread = new Thread(this.ws,
            "WorkerSender[myid=" + self.getId() + "]");
    this.wsThread.setDaemon(true);
    //創建線程WorkerReceiver
    this.wr = new WorkerReceiver(manager);
    this.wrThread = new Thread(this.wr,
            "WorkerReceiver[myid=" + self.getId() + "]");
    this.wrThread.setDaemon(true)；
}

開啓 WorkerSender 和 WorkerReceiver 線程。

WorkerSender 線程自旋獲取 sendqueue 第一層隊列元素

sendqueue 隊列元素內容爲相關選票信息詳見 ToSend 類；
首先判斷選票 sid 是否和自己 sid 值相同，相等直接放入到 recvQueue 隊列中；
不相同將 sendqueue 隊列元素轉儲到 queueSendMap<sid,queue> 第二層傳輸隊列中。

【Java】//FastLeaderElection.Messenger.WorkerSenderclass WorkerSender extends ZooKeeperThread{
//...
  public void run() {
    while (!stop) {
        try {
            ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
            if(m == null) continue;
  //將投票信息發送出去
            process(m);
        } catch (InterruptedException e) {
            break;
        }
    }
    LOG.info("WorkerSender is down");
  }
}
//QuorumCnxManager#toSend
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> oldq = queueSendMap.putIfAbsent(sid, bq);
         //轉儲到queueSendMap<sid,queue>第二層傳輸隊列中
         if (oldq != null) {
             addToSendQueue(oldq, b);
         } else {
             addToSendQueue(bq, b);
         }
         connectOne(sid);     
    }
}

WorkerReceiver 線程自旋獲取 recvQueue 第二層傳輸隊列元素轉存到 recvqueue 第一層隊列中。

【Java】
//WorkerReceiver
public void run() {
    Message response;
    while (!stop) {
      // Sleeps on receive
      try {
          //自旋獲取recvQueue第二層傳輸隊列元素
          response = manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
          if(response == null) continue;
          // The current protocol and two previous generations all send at least 28 bytes
          if (response.buffer.capacity() < 28) {
              LOG.error("Got a short response: " + response.buffer.capacity());
              continue;
          }
          //...
  if(self.getPeerState() == QuorumPeer.ServerState.LOOKING){
         //第二層傳輸隊列元素轉存到recvqueue第一層隊列中
         recvqueue.offer(n);
         //...
      }
    }
//...
}

06 選舉核心邏輯

啓動線程 QuorumPeer

開始 Leader 選舉投票 makeLEStrategy().lookForLeader()；

sendNotifications() 向其它節點發送選票信息，選票信息存儲到 sendqueue 隊列中。sendqueue 隊列由 WorkerSender 線程處理。

【plain】
//QuorunPeer.run
//...
try {
   reconfigFlagClear();
    if (shuttingDownLE) {
       shuttingDownLE = false;
       startLeaderElection();
       }
    //makeLEStrategy().lookForLeader() 發送投票
    setCurrentVote(makeLEStrategy().lookForLeader());
} catch (Exception e) {
    LOG.warn("Unexpected exception", e);
    setPeerState(ServerState.LOOKING);
}  
//...
//FastLeaderElection.lookLeader
public Vote lookForLeader() throws InterruptedException {
//...
  //向其他應用發送投票
sendNotifications();
//...
}
private void sendNotifications() {
    //獲取應用節點
    for (long sid : self.getCurrentAndNextConfigVoters()) {
        QuorumVerifier qv = self.getQuorumVerifier();
        ToSend notmsg = new ToSend(ToSend.mType.notification,
                proposedLeader,
                proposedZxid,
                logicalclock.get(),
                QuorumPeer.ServerState.LOOKING,
                sid,
                proposedEpoch, qv.toString().getBytes());
        if(LOG.isDebugEnabled()){
            LOG.debug("Sending Notification: " + proposedLeader + " (n.leader), 0x"  +
                  Long.toHexString(proposedZxid) + " (n.zxid), 0x" + Long.toHexString(logicalclock.get())  +
                  " (n.round), " + sid + " (recipient), " + self.getId() +
                  " (myid), 0x" + Long.toHexString(proposedEpoch) + " (n.peerEpoch)");
        }
        //儲存投票信息
        sendqueue.offer(notmsg);
    }
}
class WorkerSender extends ZooKeeperThread {
    //...
    public void run() {
    while (!stop) {
        try {
//提取已儲存的投票信息
            ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
            if(m == null) continue;
            process(m);
        } catch (InterruptedException e) {
            break;
        }
    }
    LOG.info("WorkerSender is down");
  }
//...
}

自旋 recvqueue 隊列元素獲取投票過來的選票信息：

【Java】
public Vote lookForLeader() throws InterruptedException {
//...
/*
 * 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
     */
    //提取投遞過來的選票信息
    Notification n = recvqueue.poll(notTimeout,
            TimeUnit.MILLISECONDS);
/*
 * Sends more notifications if haven't received enough.
 * Otherwise processes new notification.
 */
if(n == null){
    if(manager.haveDelivered()){
        //已全部連接成功，並且前一輪投票都完成，需要再次發起投票
        sendNotifications();
    } else {
        //如果未收到選票信息，manager.contentAll()自動連接其它socket節點
        manager.connectAll();
    }
    /*
     * Exponential backoff
     */
    int tmpTimeOut = notTimeout*2;
    notTimeout = (tmpTimeOut < maxNotificationInterval?
            tmpTimeOut : maxNotificationInterval);
    LOG.info("Notification time out: " + notTimeout);
         }
     //....
    }
  //...
}

【Java】
//manager.connectAll()->connectOne(sid)->initiateConnection(...)->startConnection(...)
private boolean startConnection(Socket sock, Long sid)
        throws IOException {
    DataOutputStream dout = null;
    DataInputStream din = null;
    try {
        // Use BufferedOutputStream to reduce the number of IP packets. This is
        // important for x-DC scenarios.
        BufferedOutputStream buf = new BufferedOutputStream(sock.getOutputStream());
        dout = new DataOutputStream(buf);
        // Sending id and challenge
        // represents protocol version (in other words - message type)
        dout.writeLong(PROTOCOL_VERSION);
        dout.writeLong(self.getId());
        String addr = self.getElectionAddress().getHostString() + ":" + self.getElectionAddress().getPort();
        byte[] addr_bytes = addr.getBytes();
        dout.writeInt(addr_bytes.length);
        dout.write(addr_bytes);
        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;
    }
    // authenticate learner
    QuorumPeer.QuorumServer qps = self.getVotingView().get(sid);
    if (qps != null) {
        // TODO - investigate why reconfig makes qps null.
        authLearner.authenticate(sock, qps.hostname);
    }
    // If lost the challenge, then drop the new connection
    //保證集羣中所有節點之間只有一個通道連接
    if (sid > self.getId()) {
        LOG.info("Have smaller server identifier, so dropping the " +
                "connection: (" + sid + ", " + self.getId() + ")");
        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;
}

如上述代碼中所示，sid>self.sid 纔可以創建連接 Socket 和 SendWorker,RecvWorker 線程，存儲到 senderWorkerMap<sid,SendWorker> 中。對應第 2 步中的 sid<self.sid 邏輯，保證集羣中所有節點之間只有一個通道連接。

圖 4 節點之間連接方式

【Java】
public Vote lookForLeader() throws InterruptedException {
//...
    if (n.electionEpoch > logicalclock.get()) {
        //當前選舉週期小於選票週期，重置recvset選票池
        //大於當前週期更新當前選票信息，再次發送投票
        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()) {
        if(LOG.isDebugEnabled()){
            LOG.debug("Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x"
                    + Long.toHexString(n.electionEpoch)
                    + ", logicalclock=0x" + Long.toHexString(logicalclock.get()));
        }
        break;
    } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
            proposedLeader, proposedZxid, proposedEpoch)) {//相同選舉週期
        //接收的選票與當前選票PK成功後，替換當前選票
        updateProposal(n.leader, n.zxid, n.peerEpoch);
        sendNotifications();
    }
//...
}

在上代碼中，自旋從 recvqueue 隊列中獲取到選票信息。開始進行選舉：

判斷當前選票和接收過來的選票週期是否一致
大於當前週期更新當前選票信息，再次發送投票
週期相等：當前選票信息和接收的選票信息進行 PK

【Java】
//接收的選票與當前選票PK
protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
        LOG.debug("id: " + newId + ", proposed id: " + curId + ", zxid: 0x" +
                Long.toHexString(newZxid) + ", proposed zxid: 0x" + Long.toHexString(curZxid));
        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)))));wId > curId)))));
  }

在上述代碼中的 totalOrderPredicate 方法邏輯如下：

競選週期大於當前週期爲 true
競選週期相等，競選 zxid 大於當前 zxid 爲 true
競選週期相等，競選 zxid 等於當前 zxid，競選 sid 大於當前 sid 爲 true
經過上述條件判斷爲 true 將當前選票信息替換爲競選成功的選票，同時再次將新的選票投出去。

【Java】
public Vote lookForLeader() throws InterruptedException {
//...
   //存儲節點對應的選票信息
    // key:選票來源sid  value:選票推舉的Leader sid
    recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
    //半數選舉開始
    if (termPredicate(recvset,
            new Vote(proposedLeader, proposedZxid,
                    logicalclock.get(), proposedEpoch))) {
        // Verify if there is any change in the proposed leader
        while((n = recvqueue.poll(finalizeWait,
                TimeUnit.MILLISECONDS)) != null){
            if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                    proposedLeader, proposedZxid, proposedEpoch)){
                recvqueue.put(n);
                break;
            }
        }
        /*WorkerSender
         * This predicate is true once we don't read any new
         * relevant message from the reception queue
         */
        if (n == null) {
            //已選舉出leader 更新當前節點是否爲leader 
            self.setPeerState((proposedLeader == self.getId()) ?
                    ServerState.LEADING: learningState());
            Vote endVote = new Vote(proposedLeader,
                    proposedZxid, proposedEpoch);
            leaveInstance(endVote);
            return endVote;
        }
    }
//...
}
/**
     * Termination predicate. Given a set of votes, determines if have
     * sufficient to declare the end of the election round.
     *
     * @param votes
     *            Set of votes
     * @param vote
     *            Identifier of the vote received last  PK後的選票
     */
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());
    }
    /*
     * First make the views consistent. Sometimes peers will have different
     * zxids for a server depending on timing.
     */
    //votes 來源於recvset 存儲各個節點推舉出來的選票信息
    for (Map.Entry<Long, Vote> entry : votes.entrySet()) {
//選舉出的sid和其它節點選擇的sid相同存儲到voteSet變量中。
        if (vote.equals(entry.getValue())) {
//保存推舉出來的sid
            voteSet.addAck(entry.getKey());
        }
    }
    //判斷選舉出來的選票數量是否過半
    return voteSet.hasAllQuorums();
}
//QuorumMaj#containsQuorum
public boolean containsQuorum(Set<Long> ackSet) {
    return (ackSet.size() > half);
   }

在上述代碼中：recvset 是存儲每個 sid 推舉的選票信息。

第一輪 sid1:vote(1,0,1) ,sid2:vote(2,0,1)；

第二輪 sid1:vote(2,0,1) ,sid2:vote(2,0,1)。

最終經過選舉信息 vote(2,0,1) 爲推薦 leader, 並用推薦 leader 在 recvset 選票池裏比對持相同票數量爲 2 個。因爲總共有 3 個節點參與選舉，sid1 和 sid2 都選舉 sid2 爲 leader，滿足票數過半要求，故確認 sid2 爲 leader。

setPeerState 更新當前節點角色；
proposedLeader 選舉出來的 sid 和自己 sid 相等，設置爲 Leader；
上述條件不相等，設置爲 Follower 或 Observing；
更新 currentVote 當前選票爲 Leader 的選票 vote(2,0,1)。

07 總結

通過對 Leader 選舉源碼的解析，可以瞭解到：

多個應用節點之間網絡通信採用 BIO 方式進行相互投票，同時保證每個節點之間只使用一個通道，減少網絡資源的消耗，足以見得在 BIO 分佈式中間件開發中的技術重要性。
基於 BIO 的基礎上，靈活運用多線程和內存消息隊列完好實現多層隊列架構，每層隊列由不同的線程分工協作，提高快速選舉性能目的。
爲 BIO 在多線程技術上的實踐帶來了寶貴的經驗。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/RvXOhy7O9Zx99klzfUoOuA

導讀