ZAB协议恢复模式-leader选举

之前在网上看了很多zookeeper的zab原理，但讲述的都不够详细，很多地方模糊不清，所以就研究了一下zookeeper源码，并整理成几篇博客，希望对其他小伙伴有所帮助。本文是ZAB协议崩溃恢复Leader选举部分的内容，数据同步见另一篇博客 《ZAB协议恢复模式-数据同步》     。

为了避免理解上的歧义，将投票动作和投票信息区分开，在本文中，我将服务器的投票信息称之为选票。

一  基本概念

1.  Noitifcation

Notification其实是选举过程中的通信信息；选举过程主要围绕Notification进行。

选Leader过程中Zookeeper server（QuorumPeer）都会根据Notification信息生成Vote（选票信息）。为了方便以下理解，我们不妨将Notification看成每个zookeeper server的选票信息。

以下是notification的主要字段。

1)   zxid

事务id，事务请求的唯一标记，由leader服务器负责进行分配。

高32位是peerEpoch，低32位是请求的计数，从0开始。

2)   peerEpoch

每次leader选举完成之后，都会选举出一个新的peerEpoch，用来标记事务请求所属的轮次。

3)   electionEpoch

每次leader选举，electionEpoch就会自增1，统计选票信息时，首先保证electionEpoch相同。

4)   sid

服务器id

5)   leader

提议的leader。

 

2.  其他概念

1)   lastProcessedZxid

最后一次commit的事务请求的zxid

 

二  Zookeeper Server

1.  QuorumPeer

Zookeeper server启动始于QuorumPeerMain#main()。

Zookeeper server的主要逻辑都在QuorumPeer中，此类具体负责的逻辑如下流程图：

本文仅介绍Leader选举流程内容，其他流程（follower流程、leader流程、observer流程）见《ZAB协议恢复模式-数据同步》。

从流程图中可以看到，QuorumPeer将一直运行，直到running=false。While无限循环中，根据当前zookeeper服务器的投票状态进入不同的业务逻辑。

服务器启动时处于LOOKING状态；退出任何子流程以后状态立即被改成LOOKING状态。LOOKING状态，表示Zookeeper服务端在进行选举流程。

在集群环境下，任何一台服务器都可能被选中成为leader，但每台服务器成为leader的可能性会有所不同，具体为：zxid、peerEpoch、electionEpoch、sid大者更容易被选举为leader，选举流程部分会详细讲述此中缘由。

 

2.  状态

1)   LOOKING

表示在选举leader。

2)   FOLLOWING

服务器的角色为follower。

3)   LEADING

服务器的角色为leader。

4)   OBSERVING

服务器的角色为observer，此种服务器不参与投票，只是同步状态。

 

三  Leader选举

1.  选举须知

选举流程比较复杂，在正式进入选举流程之前，需要先弄清楚以下内容：

每个Zookeeper服务端进入LOOKING状态以后，都会发起选举流程，默认情况下是快速选举，所以由FastLeaderElection#lookForLeader方法承担此职责。

每个Zookeeper服务器接收到选票提议以后，只有两个选择：

• 接受选票提议，认可提议中推荐的服务器作为Leader候选人；
  
• 不接受选票提议，推荐自己上一次推荐的服务器作为Leader候选人。（选举开始是总是推荐自己作为候选人，选举中会根据收到的选票信息决定是否更换推荐候选人）
  

默认情况下，至少超过半数(即n/2+1)服务器投票给同一个Leader候选人时，Leader候选人才有可能被选中为Leader。（这里说的是有可能，还需要进行一些其他逻辑进行验证）。

 

2.  流程

以上为代码的完整流程，看起来比较复杂，我们可以按照以下内容简单理解一下。

需要说明的是：此流程结束仅仅是确认那个服务器成为Leader，具体Leader是否能够最终成为Leader，还有另外的流程决定，这部分内容会见《ZAB协议恢复模式-数据同步》。

 

3.  流程详述

流程比较复杂，接下来对流程图中标有数字的地方详细介绍。

1)   1-自增logicalclock

Logicalclock就是Notification中的electionEpoch.

选举的第一个操作是logicalclock自增，接着更新提议信息，其实第一次总是提议自己作为Leader。

如果和现实中总统选举做一个类比的话，每次总统选举时都要明确这是第几届选举，logicalclock就对应的是“第几届”。整个选举必须保证处于同一届选举中方有效。

 

2)   2-发送选票信息

这是一个异步操作（由sendNotifications封装），将提议信息放到FastLeaderElection#sendqueue队列中，然后异步的发送个所有其他zookeeper server（这里指的是所有参与投票的服务器，不会发送个Observer类型的服务器）。

 

3)   3-从选票队列中取选票信息

当前server收到其他服务器的选举回复信息以后，将选票信息放在FastLeaderElection#recvqueue，当服务器循环从此队列中取选票信息。

如果队列中有选票信息立即返回，如果没有则等待。这里有一个超时时间，如果超过此时间依然没有选票信息，则返回null，这么做可以防止死等。

 

4)   4-判断消息是否发送出去

当从recvqueue没有取得选票信息时，会检查是否已经将提议的leader发送给其他server了，如果queueSendMap（待发送队列）为空，说明已经全部发送出去了；否则认为没有发送出去，此时会重连其他zookeeper server，保证链路畅通。

 

5)   5-重连其他ZookeeperServer

如果链路出现异常，可能会导致提议信息无法发送成功，所以如果queueSendMap中的信息没有全部发送出去，此时会重连其他zookeeperserver，以保证zookeeper集群的链路畅通。

 

6)   6-LOOKING状态时，electionEpoch比较

如果收到的选票信息状态为LOOKING，说明对方也在选举中。

a)   electionEpoch比较

进行electionEpoch比较的目的是统一当前是第几届选举。

• 如果收到选票的electionEpoch更大，那么使用收到选票的electionEpoch作为“届”，然后清空收到的选票信息，更新提议信息（这里有一个判断过程），重新发送更新后的提议。
  
• 如果收到选票的electionEpoch小，直接忽略此选票。
  
• 如果收到选票的electionEpoch和当前相同，那么认为是合法的选票，接着判断是否应该更新选票。
  

b)   接受选票的提议

当且仅当以下三个条件满足其一时，将接受选票的提议，并重新发送选票信息。

• n.peerEpoch > self.proposedEpoch
  
• n.peerEpoch == self.proposedEpoch&& n.zxid > self.proposedZxid
  
• n.peerEpoch == self.proposedEpoch&& n.zxid = self.proposedZxid && n.leader > self.proposedLeader
  

n指的是收到的选票，self指的是当前服务器自身的提议

由此可知：peerEpoch、zxid、leader越大，与有可能成为Leader。

proposedLeader开始的时候一定是当前server的id，但随着选举的进行，会变成上一次提议的leader。

 

7)   7-Leader是否有效

如果某一个server已经得到半数以上的选票，那么进入leader是否有效的验证逻辑，具体如下：

无限循环的从recvqueue中取选票，满足一下条件之一时退出循环：

• recvqueue没有选举票（超时时间内一直没有获取到选票）；
  
• 取到一个更新的选票信息（满足“接受选票提议”的条件，则说明提议更新）。
  

这里其实是一个Leader有效性的校验。依次从recvqueue中取出所有的选票，校验发现所有的选票均满足“接受选票提议”时，说明没有服务器的选票能够推翻之前的结论，所以此时可以认为Leader是有效的。

 

8)   8-FOLLOWING、LEADING状态时，electionEpoch比较

a)   选票集合

为了将此部分解释清楚，需要先能清楚选举过程中用到的两个集合：

recvset：用来记录选票信息，以方便后续统计;

outofelection：用来记录选举逻辑之外的选票，例如当一个服务器加入zookeeper集群时，因为集群已经存在，不用重新选举，只需要在满足一定条件下加入集群即可。

 

b)   electionEpoch比较

如果收到的选票显示处于FOLLOWING、LEADING状态，说明集群目前有Leader，只需要确保当前服务器和Leader能够正常通信，并收到了集群半数以上服务器推荐推荐此Leader时，就直接加入到集群中去。

因为Leader已经存在，所以所有的选票都会加入到outofelection中。如果outofelection有一条选票是来自leader的，那么就可以认为自己和leader正常通信；如果outofelection中统计出有超过半数的服务器都推荐了这个leader，那么毫无疑问，此选票推荐的就是我们的leader。

 

c)   源码逻辑

• 如果logicalclock = n.electionEpoch相同，那么将此选票加入到选票列表中，如果此张选票通过了“选票有效性验证”，那么将此选票推举的候选人作为leader。
  
• 因为leader已存在，将所有选票放在outofelection中，进行一次选票“选票有效性验证”，如果通过就可以将次选票推举的候选人作为leader。
  

以上两步的差别是在进行有效性校验时，一个用的是recvset，一个用的是outofelection。从代码上看，zookeeper认为只要electionEpoch就认为这是在选举，所以判断选票数目的时候使用的是recvset。

以上两步逻辑比较绕，如果理解起来比较困难，可以参考一下源码。

 

d)   源码

case FOLLOWING:
case LEADING:
if (n.electionEpoch == logicalclock) {
   //如果Notification的electionEpoch和当前的electionEpoch相同，那么说明在同一轮的选举中，
   recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
 
   //判定选举是否结束
    if (ooePredicate(recvset,outofelection, n)){
       // 选举结束，设置状态
       self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());
 
       Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
       leaveInstance(endVote);
       return endVote;
    }
}
 
outofelection.put(n.sid,
       new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
// 判定选举是否结束
if(ooePredicate(outofelection, outofelection, n)) {
   synchronized (this) {
       logicalclock = n.electionEpoch;
       self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());
    }
   Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
   leaveInstance(endVote);
   return endVote;
}
break;

 

e)   选票有效性验证

• leader候选人获得超过半数的选票。
  
• 通过Leader有效性校验。
  

 

f)   Leader有效性验证

• 如果自己不是leader，那么一定要收到过Leader的信息，即收到Leader信息，并且leader的回复信息中宣称自己的状态是ServerState.LEADING
  
• 如果自己是leader，那么当前logicalclock一定要等于选票信息中的electionEpoch
  

 

4.  核心类

以下为选举过程中使用到的核心类。

1)   QuorumPeer

见整理流程部分

2)   FastLeaderElection

默认的选举算法，即上面流程图描述的内容。

此类有几个重要的内部类，如下：

a)   FastLeaderElection#Messenger#WorkerReceiver

从QuorumCnxManager#recvQueue中获取网络包，并将其发到FastLeaderElection#recvqueue中

b)   FastLeaderElection#Messenger#WorkerSender

从FastLeaderElection#sendqueue中获取网络包，并将其放到QuorumCnxManager#queueSendMap中，并发送到网络上

 

3)   QuorumCnxManager

QuorumCnxManager是实际发生网络交互的地方。QuorumCnxManager保证与每一个zookeeper服务器之间只有一个链接。

主要数据结构如下：

a)   queueSendMap

sid（key） -> buffer queue（value），为每个参与投票的server都保留一个队列。

b)   recvQueue

message queue，所有收到的消息都放到recvQueue。

c)   listener

server主线程，收发消息时和上面两个队列交互。

 

四  源码分析

以下是选举过程中用到的主要代码以及注释。代码版本为zookeeper-3.4.10。

我给选举逻辑的部分代码加了注释，以便于理解，下载地址：git@gitee.com:wuzhengfei/zookeeper-3.4.10-sources.git

 

1.  FastLeaderElection

1)   lookForLeader

   public Vote lookForLeader() throws InterruptedException {
       try {
           self.jmxLeaderElectionBean = newLeaderElectionBean();
           MBeanRegistry.getInstance().register(self.jmxLeaderElectionBean,self.jmxLocalPeerBean);
       } catch (Exception e) {
            LOG.warn("Failedto register with JMX", e);
            self.jmxLeaderElectionBean = null;
       }
       if (self.start_fle== 0) {
           self.start_fle = System.currentTimeMillis();
       }
       try {
           // 投票信息列表
           HashMap<Long, Vote> recvset = newHashMap<Long, Vote>();
 
           HashMap<Long, Vote> outofelection = newHashMap<Long, Vote>();
 
           int notTimeout = finalizeWait;
 
           synchronized(this) {
               //更新选举周期
               logicalclock++;
                // 更新提议信息，提议自己作为leader，
               updateProposal(getInitId(),getInitLastLoggedZxid(), getPeerEpoch());
           }
 
           LOG.info("New election. My id =  " + self.getId() + ",proposed zxid=0x" + Long.toHexString(proposedZxid));
           // 发送选票信息
           sendNotifications();
 
           /*
             * 循环直到选举出leader
             */
           while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){
                // 从recvqueue移除第一个Notification（选票信息），最多等待notTimeout ms
               Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);
 
                // 如果当前没有收到足够多的Notification，不足以产生leader，那么继续发送选票信息，否则继续处理收到的Notification
               if(n == null){
                   // 当前没有收到回复信息
                   if (manager.haveDelivered()){
                       // 如果消息全部投递出去了，那么在发送一次选票信息
                      sendNotifications();
                   } else {
                       // 如果消息没有投递出去，那么尝试重连
                       manager.connectAll();
                   }
 
                   int tmpTimeOut= notTimeout * 2;
                   notTimeout = (tmpTimeOut < maxNotificationInterval ? tmpTimeOut : maxNotificationInterval);
                   LOG.info("Notificationtime out: " + notTimeout);
               } elseif (self.getVotingView().containsKey(n.sid)) {
                   /*
                    * Only proceed if the vote comes from a replica in the
                    * voting view.
                    */
                   switch (n.state) {
                   case LOOKING:
                       // If notification > current, replace and sendmessages
                       // out
                       if (n.electionEpoch > logicalclock){
                          logicalclock = n.electionEpoch;
                          // 清空收到的选票信息列表
                          recvset.clear();
                           // 比较notification，current，大的最为下次推荐的leader，然后发送选票信息
                          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) {
                           // 如果notification#electionEpoch小,认为是无效选票，直接忽略
                          if (LOG.isDebugEnabled()){
                              LOG.debug(
                                       "Notificationelection epoch is smaller than logicalclock. n.electionEpoch = 0x"
                                             + Long.toHexString(n.electionEpoch) + ",logicalclock=0x"
                                             + Long.toHexString(logicalclock));
                          }
                          break;
                       } else if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader,proposedZxid,
                              proposedEpoch)){
                           // 比较notification，current，大的最为下次推荐的leader，然后发送选票信息
                          updateProposal(n.leader, n.zxid, n.peerEpoch);
                          sendNotifications();
                       }
 
                       if (LOG.isDebugEnabled()){
                          LOG.debug("Adding vote: from=" + n.sid + ", proposed leader=" + n.leader
                                  + ",proposed zxid=0x" + Long.toHexString(n.zxid) + ",proposed election epoch=0x"
                                  + Long.toHexString(n.electionEpoch));
                       }
 
                       // 将收到的选票加入到选票列表中
                       recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch,n.peerEpoch));System.err.println("-----"+DateUtil.formatTime(new Date(), DateUtil.PATTERN_TIMESTEMP_SSS)+"    recvset="+JSON.toJSONString(recvset));
                      System.err.println("-----"+DateUtil.formatTime(new Date(), DateUtil.PATTERN_TIMESTEMP_SSS)+"    recvset="+JSON.toJSONString(recvset));
                       
                       // 判断self提议的服务器是否已经获得超过半数的票
                       if (termPredicate(recvset,
                               new Vote(proposedLeader, proposedZxid, logicalclock, proposedEpoch))) {
 
                           // 循环直到当recvqueue中没有Notification，或有出现一个Notification的选票信息比推举的leader更大时，方跳出循环
                          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) {
                               // 进入这里说明recvqueue中没有比推举leader更大的票，可以将推举的服务器作为leader了。
                              self.setPeerState(
                                      (proposedLeader == self.getId())? ServerState.LEADING : learningState());
 
                              Vote endVote = new Vote(proposedLeader,proposedZxid, logicalclock,proposedEpoch);
                              leaveInstance(endVote);
                              return endVote;
                          }
                       }
                       break;
                   case OBSERVING:
                       LOG.debug("Notificationfrom observer: " + n.sid);
                       break;
                   case FOLLOWING:
                   case LEADING:
                       /**
                        * if内的代码、if外的代码比较类似，唯一的差别是ooePredicate方法的第一个参数，
                        * 考虑集群中只有leader存活，其他所有服务器都处于LOOKING状态，就能理解为什么需要这两段逻辑了。
                        */
                       if (n.electionEpoch == logicalclock){
                           // 如果Notification的electionEpoch和当前的electionEpoch相同，那么说明在同一轮的选举中
                          recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
 
                          // 判定选举是否结束
                          if (ooePredicate(recvset, outofelection,n)) {
                              // 选举结束，设置状态
                              self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());
 
                              Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
                              leaveInstance(endVote);
                              return endVote;
                          }
                       }
 
                       outofelection.put(n.sid,
                              new Vote(n.version, n.leader, n.zxid, n.electionEpoch,n.peerEpoch,n.state));
                       // 判定选举是否结束
                       if (ooePredicate(outofelection,outofelection, n)){
                          synchronized (this) {
                              logicalclock = n.electionEpoch;
                              self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());
                          }
                          Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
                          leaveInstance(endVote);
                          return endVote;
                       }
                       break;
                   default:
                       LOG.warn("Notification state unrecognized:{} (n.state), {} (n.sid)", n.state, n.sid);
                       break;
                   }
               } else{
                   LOG.warn("Ignoring notification from non-clustermember " + n.sid);
               }
           }
           return null;
       } finally {
           try {
               if(self.jmxLeaderElectionBean!= null) {
                  MBeanRegistry.getInstance().unregister(self.jmxLeaderElectionBean);
               }
           } catch(Exception e) {
                LOG.warn("Failedto unregister with JMX", e);
           }
            self.jmxLeaderElectionBean = null;
           LOG.debug("Number of connection processing threads: {}",manager.getConnectionThreadCount());
       }
   }

 

2)   totalOrderPredicate

   /**
     * 如果new票信息大于当前的票信息，那么返回true。以下三种case，只要满足一种就返回true：
     * <li>newEpoch > curEpoch
     * <li>newEpoch == curEpoch&& newZxid > curZxid
     * <li>newEpoch == curEpoch&& newZxid = curZxid && newId > curId
     */
   protected booleantotalOrderPredicate(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 thefollowing three cases hold: 1- New epoch
         * is higher 2- New epoch isthe same as current epoch, but new zxid is
         * higher 3- New epoch is thesame 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)))));
   }

 

3)   termPredicate

   /**
     * 是否终止投票，如果{@linkvote}对应的服务器获得了超过半数的票，那么返回true
     * 
     * @param votes
     * @param vote
     * @return
     */
   protected booleantermPredicate(HashMap<Long, Vote> votes,Vote vote) {
 
       HashSet<Long> set= new HashSet<Long>();
 
       /*
         * First make the viewsconsistent. Sometimes peers will have different
         * zxids for a serverdepending on timing.
         */
       for (Map.Entry<Long,Vote> entry : votes.entrySet()){
           if (vote.equals(entry.getValue())){
               set.add(entry.getKey());
           }
       }
 
       return self.getQuorumVerifier().containsQuorum(set);
   }

 

4)   checkLeader

   /**
     * leader确认检查
     * <li>如果自己不是leader，那么一定要收到过Leader的信息，并且leader宣称自己的状态是ServerState.LEADING；
     * <li>如果自己是leader，那么当前logicalclock一定要等于选票信息中的electionEpoch
     * 
     * @param outofelection
     * @param leader
     * @param electionEpoch
     */
   protected booleancheckLeader(HashMap<Long, Vote> outofelection,long leader,long electionEpoch){
 
       boolean predicate = true;
 
       /*
         * If everyone else thinks I'mthe leader, I must be the leader. The
         * other two checks are justfor the case in which I'm not the leader.
         * If I'm not the leader and Ihaven't received a message from leader
         * stating that it is leading,then predicate is false.
         */
 
       if (leader != self.getId()){
           if (outofelection.get(leader)== null)
                // 表示还未从leader中收到任何信息，所以选举结束未结束。
               predicate= false;
           else if (outofelection.get(leader).getState() != ServerState.LEADING)
                // 指推举的那个leader还没有将自己的状态改成leader，所以选举依然未完成。
               predicate= false;
       } else if (logicalclock!= electionEpoch) {
            // 自己是leader时，这两个值应该是相同的，如果不同说明有异常，不能结束选举。
           predicate= false;
       }
 
       return predicate;
   }

 

5)   ooePredicate

   /**
     * 是否能宣称选举结束
     * <li>{@code n}获得了半数以上的票
     * <li>通过了leader确认的检查
     * 
     * @param recv
     *           map of received votes
     * @param ooe
     *           map containing out of election votes (LEADING or FOLLOWING)
     * @param n
     *           Notification
     * @return
     */
   protected booleanooePredicate(HashMap<Long, Vote> recv,HashMap<Long, Vote> ooe, Notification n) {
 
       return (termPredicate(recv, new Vote(n.version, n.leader, n.zxid, n.electionEpoch,n.peerEpoch,n.state))
               && checkLeader(ooe, n.leader, n.electionEpoch));
 
   }

 

五  选举示例分析

假设zookeeper集群的部署信息如下：

server.1=192.168.1.1:2888:3888

server.2=192.168.1.2:2888:3888

server.3=192.168.1.3:2888:3888

1.  集群启动时

根据上面的信息，由此我们可以得出集群启动时具有如下信息：

Zookeeper每一台服务器启动时，默认状态为LOOKING状态，所以都会启动选leader流程。当然如果只有一台启动成功，无法满足选举的超过一半的条件，所以永远无法完成选leader逻辑。

 

1)   S1服务器选举过程

为了描述的简单，假设集群只启动S1、S2服务器，我们看一下选举过程是怎样的。

a)   S1发送提议

S1启动后首先就是进行Leader选举。S1首先logicalclock改为1，接着发送选票信息，S1的选票信息如下：

b)   S2收到S1选举提议

S2启动后就会进入选举流程，因为logicalclock=1，所以忽略同步时钟这个操作；此时S2的提议信息为：

S2收到信息以后，将消息放在recvqueue中，接着会比较S1的提议和自己的提议信息，此时S2发现self.proposedLeader>S1.proposedLeader，所以S2回复提议信息如下：

c)   S1收到S2回复的选票信息

S1比较提议后发现自己推举的Leader小，所以将自己的提议信息更新为：

d)   S2再次收到S1的选票信息

此时S2不用更新提议信息，直接回复S1，回复的提议内容如下：

e)   S1收到S2的回复的选举提议

此时S1自己的选票信息为：

收到S2的选票信息为：

汇总后S1发现，目前收到的选票信息如下：

因为满足超过半数的条件，所以S1会认为S2是leader，S1退出选举流程。

2)   S1服务器收到的选票信息

a)   S1未收到S2的选票信息

recvset={1:{"electionEpoch":1,"id":1,"peerEpoch":0,"state":"LOOKING","version":0,"zxid":0}}

 

b)   S1收到S2的选票信息

recvset={1:{"electionEpoch":1,"id":1,"peerEpoch":0,"state":"LOOKING","version":0,"zxid":0},2:{"electionEpoch":1,"id":2,"peerEpoch":0,"state":"LOOKING","version":0,"zxid":0}}

进行到这一步是S1会更新自己的提议信息，改成选举S2作为leader

 

c)   S1收到S2的选票信息

recvset={1:{"electionEpoch":1,"id":2,"peerEpoch":0,"state":"LOOKING","version":0,"zxid":0},2:{"electionEpoch":1,"id":2,"peerEpoch":0,"state":"LOOKING","version":0,"zxid":0}}

进行到这一步就可以认为S2服务器当选成为Leader了。

 

3)   S2的leader选举

S2的选举过程和上面基本一样，唯一不同的地方是：S2一直选举的都是S2服务器，所以不用像S1一样选举过程中修改提议的Leader。

 

 

2.  新服务器加入集群

a)   初始条件假设

继续上面的示例，假设集群开始只启动了S1、S2服务器，运行一段时间以后S3服务器希望加入到集群中。

假设运行一段时间以后集群的情况如下：

S3新加入集群，加入时的情况如下：

b)   S3发送提议

S3启动后会发起选主流程，发送的提议信息如下：

c)   S1收到S3的提议信息

因为S3已经是Follower，并且epoch、zxid均大于S3，所以回复的提议如下：

d)   S2收到S3的提议信息

因为S2已经是Leader，并且epoch、zxid均大于S3，所以回复的提议如下：

 

e)   S3收到回复信息

S3统计选票信息会发现S1、S2都推荐S2作为Leader,并且S2已经是leader，因为超过半数推荐S2，所以此时S3也会将S2作为leader，结束选主流程

 

3.  又有新服务器加入集群

继续假设按照上面信息部署集群，又来了一个服务器S4希望加入集群，那么流程会如何呢？

选举过程中存在以下代码：

if (self.getVotingView().containsKey(n.sid)) {

    // 选主逻辑

}

这意味着S1、S2、S3收到S4的选票请求，认为这不是当前集群的服务器，直接忽略选票信息。结论是：S4无法加入集群的。

 

4.  集群运行中重新选举

运行过程中如果Leader失效，或者超过半数服务器与Leader不同步等等情况都可能导致集群运行一段时间后，触发了选主流程。接下来我们以一种最为复杂的方式来介绍选举的完整过程。假设集群信息如下：

Zxid的高位是peerEpoch，低位是自增的序列，所以如果peerEpoch大，那么zxid一定大。为了方便描述，zxid我改成“高位_地位”的形式了。

说明：一般而言，正常运行过程是不会出现以上假设的数据的，除非数据被人修改过，或者其他未知原因导致所有服务器的数据均错乱了。

因为集群中没有了leader，所以所有服务器均将重新进入选主流程。接下来我们就以S3服务器为例来讨论选主流程。

a)   第一轮选票

S3发出的选票信息信息为：

S2收到提议后回复的提议信息如下：

S1收到提议后回复的提议信息如下：

b)   第一轮选票汇总

假设S3服务器先收到S2的回复，因为S3.proposedEpoch < S2.proposedEpoch，所以S3先更新提议的信息，然后重新发送提议(P1)，新的提议如下：

紧接着接着S3又收到S1的回复，又发现S3.proposedEpoch < S1.proposedEpoch，所以S3再一次更新提议信息，然后发送新的提议(P2)，新的提议如下：

因为提议P2最终会覆盖提议P1，所以我们直接忽略S1、S2对P1的回复。直接讨论他们对P2的回复。

S1收到选票的提议以后，回复以下提议信息：

S2收到选票的提议以后，回复以下提议信息：

 

c)   第二轮选票汇总

S3收到的提议信息如下：

S3汇总以后得出的结论是S1成为Leader，然后退出选举。