摘要

这一节讲解leader选举算法源码分下，主要讲解

相关概念，定义介绍
  服务器状态
  投票
内部类
  Notification:包装接收到的数据
  ToSend:包装发送的数据
  Messenger#WorkerReceiver:线程，不断接受其他其他server消息进行处理
  Messenger#WorkerSender:线程，不断从发送队列获取待发送的消息，进行发送
属性
函数
  构造函数
  启动相关函数
  信息获取相关函数
  选举相关函数
    totalOrderPredicate：比较两张票谁能赢
    termPredicate:验证是否通过集群验证器的验证(通常是过半)
    checkLeader:接收某些状态已经为leading和following的消息时,判断leader的有效性
    ooePredicate:接收某些状态已经为leading和following的消息时,判断自己是否可以加入这个集群
    lookForLeader:选举leader的函数，根据接收消息的不同state，进行竞选周期比较，过半验证等等
思考

内容较多，走马观花的话，可以直接看图，或者直接看选举相关函数部分

概念介绍

先简单介绍一些概念

服务器状态

QuorumPeer.ServerState类

　　LOOKING：寻找Leader状态。当服务器处于该状态时，它会认为当前集群中没有Leader，因此需要进入Leader选举状态。
　　FOLLOWING：跟随者状态。表明当前服务器角色是Follower。
　　LEADING：领导者状态。表明当前服务器角色是Leader。
　　OBSERVING：观察者状态。表明当前服务器角色是Observer。

投票

Vote类

　　id：被推举的Leader的SID。
　　zxid：被推举的Leader事务ID。
　　electionEpoch：逻辑时钟，用来判断多个投票是否在同一轮选举周期中，该值在服务端是一个自增序列，每次进入新一轮的投票后，都会对该值进行加1操作。
　　peerEpoch：被推举的Leader的epoch。
　　state：当前服务器的状态。

内部类

FastLeaderElection内部类

即Notification,ToSend,Messenger三种，介绍如下

Notification

这个类用于包装接收到的数据，属性如下

Notification属性

和上面讲的Vote基本一样，不过CURRENTVERSION和version就相当于标记代码版本，用于一些向前兼容的逻辑

ToSend

这个类用于包装发送的数据，属性如下

ToSend属性

也和上面的Vote基本一样，多的一个sid代表要发给哪个server

Messenger

分为WorkerReceiver和WorkerSender两个子类

WorkerReceiver

作为接收消息的类,线程完成接收消息，然后针对性的完成如下工作

    recvqueue队列的添加
    sendqueue队列的添加

源码太长了就不贴出来了，主要流程如下

WorkerReceiver线程流程图

WorkerSender

这个代码就很简单了，主要方法如下

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

            /**
             * Called by run() once there is a new message to send.
             *
             * @param m     message to send
             */
            void process(ToSend m) {
                ByteBuffer requestBuffer = buildMsg(m.state.ordinal(), 
                                                        m.leader,
                                                        m.zxid, 
                                                        m.electionEpoch, 
                                                        m.peerEpoch);
                manager.toSend(m.sid, requestBuffer);
            }

就是不断的从sendQueue中从sendQueue取出ToSend对象，代表要发送的消息，然后构建ByteBuffer发送给对应sid的server

属性

FastLeaderElection属性

介绍如下

注意:
proposedEpoch是提议leader的epoch而不是提议leader的选举周期

函数

FastLeaderElection函数

针对主要函数进行讲解

构造函数

    public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager){
        this.stop = false;
        this.manager = manager;//连接管理器
        starter(self, manager);
    }

两个参数QuorumPeer是当前投票者，QuorumCnxManager 是选举leader时的网络IO管理器

启动相关函数

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

就是初始化当前vote以及两个队列，并且启动WorkerSender和WorkerReceiver两个线程

投票相关函数

分为

更新投票字段
生成投票的函数

updateProposal

更新投票的字段

    synchronized void updateProposal(long leader, long zxid, long epoch){
        if(LOG.isDebugEnabled()){
            LOG.debug("Updating proposal: " + leader + " (newleader), 0x"
                    + Long.toHexString(zxid) + " (newzxid), " + proposedLeader
                    + " (oldleader), 0x" + Long.toHexString(proposedZxid) + " (oldzxid)");
        }
        proposedLeader = leader;
        proposedZxid = zxid;
        proposedEpoch = epoch;
    }

getVote

生成投票的函数

    synchronized Vote getVote(){
        return new Vote(proposedLeader, proposedZxid, proposedEpoch);
    }

信息获取相关函数

learningState获取当前server角色

    private ServerState learningState(){
        if(self.getLearnerType() == LearnerType.PARTICIPANT){
            LOG.debug("I'm a participant: " + self.getId());
            return ServerState.FOLLOWING;
        }
        else{
            LOG.debug("I'm an observer: " + self.getId());
            return ServerState.OBSERVING;
        }
    }

getInitId返回当前server的sid

    private long getInitId(){//如果是参与者角色，返回myid
        if(self.getLearnerType() == LearnerType.PARTICIPANT)
            return self.getId();
        else return Long.MIN_VALUE;
    }

getInitLastLoggedZxid获取当前server的lastZxid

    private long getInitLastLoggedZxid(){//获取lastZxid
        if(self.getLearnerType() == LearnerType.PARTICIPANT)
            return self.getLastLoggedZxid();
        else return Long.MIN_VALUE;
    }

getPeerEpoch获取当前server的epoch

    private long getPeerEpoch(){//获取epoch号
        if(self.getLearnerType() == LearnerType.PARTICIPANT)
            try {
                return self.getCurrentEpoch();
            } catch(IOException e) {
                RuntimeException re = new RuntimeException(e.getMessage());
                re.setStackTrace(e.getStackTrace());
                throw re;
            }
        else return Long.MIN_VALUE;
    }

选举相关函数

totalOrderPredicate

函数比较两个投票中，哪张投票能"赢"

    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){//验证器拿到newId这个server的权限为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)))));//新的epoch大，或者同一个epoch但是新的zxid大，或者新的serverid比旧的大
    }

简而言之就是投票新的赢，投票一样新则sid大的赢，参考下面"思考"部分

termPredicate

验证自己的投票是否通过了集群验证器的验证(通常是过半)

    protected boolean termPredicate(//判断是否投票通过了集群验证器的验证
            HashMap<Long, Vote> votes,
            Vote vote) {

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

        /*
         * First make the views consistent. Sometimes peers will have
         * different zxids for a server depending on timing.
         */
        for (Map.Entry<Long,Vote> entry : votes.entrySet()) {
            if (vote.equals(entry.getValue())){
                set.add(entry.getKey());
            }
        }

        return self.getQuorumVerifier().containsQuorum(set);
    }

参数vote是自己的投票，votes是收到的投票提议集合
整个流程就是 投票集合中，投的票和自己票一样的筛选出来，看是否过半

checkLeader

该函数用于接收某些状态已经为leading和following的消息时(即部分server认为leader已经选举出来了，自己还在looking状态下),检验leader的有效性

    protected boolean checkLeader(
            HashMap<Long, Vote> votes,
            long leader,
            long electionEpoch){

        boolean predicate = true;

        /*
         * If everyone else thinks I'm the leader, I must be the leader.
         * The other two checks are just for the case in which I'm not the
         * leader. If I'm not the leader and I haven't received a message
         * from leader stating that it is leading, then predicate is false.
         */

        if(leader != self.getId()){// 自己不为leader
            if(votes.get(leader) == null) predicate = false;// 投票记录中，没有来自leader的投票
            else if(votes.get(leader).getState() != ServerState.LEADING) predicate = false;//leader不知道自己是leader
        } else if(logicalclock != electionEpoch) {// 如果大家认为我是leader，但是逻辑时钟不等于选举周期
            predicate = false;
        } 

        return predicate;
    }

ooePredicate

该函数用于接收某些状态已经为leading和following的消息时(即部分server认为leader已经选举出来了，自己还在looking状态下),自己是否可以加入这个集群

    protected boolean ooePredicate(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));
        
    }

就是过半验证，再加上leader有效性验证

lookForLeader

这就是最重要的函数，选举或者寻找leader
先贴源码以及解释

    public Vote lookForLeader() throws InterruptedException {
        try {
            self.jmxLeaderElectionBean = new LeaderElectionBean();
            MBeanRegistry.getInstance().register(
                    self.jmxLeaderElectionBean, self.jmxLocalPeerBean);
        } catch (Exception e) {
            LOG.warn("Failed to register with JMX", e);
            self.jmxLeaderElectionBean = null;
        }
        if (self.start_fle == 0) {
           self.start_fle = System.currentTimeMillis();//开始fast leader election的时间
        }
        try {
            HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();//本轮次logicalclock, 接收到的投票集合

            HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();//选举之外的投票集合(即对方为following和leading状态)

            int notTimeout = finalizeWait;//notify timeout

            synchronized(this){
                logicalclock++;//轮次+1
                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());//更新提议，包含(myid,lastZxid,epoch)
            }

            LOG.info("New election. My id =  " + self.getId() +
                    ", proposed zxid=0x" + Long.toHexString(proposedZxid));
            sendNotifications();//给其他参与者发送自己的提议

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

            while ((self.getPeerState() == ServerState.LOOKING) &&
                    (!stop)){//只要没有停止，并且状态处于LOOKING状态
                /*
                 * Remove next notification from queue, times out after 2 times
                 * the termination time
                 */
                Notification n = recvqueue.poll(notTimeout,
                        TimeUnit.MILLISECONDS);//在200ms内接收通知

                /*
                 * Sends more notifications if haven't received enough.
                 * Otherwise processes new notification.
                 */
                if(n == null){
                    if(manager.haveDelivered()){//如果已经发送过
                        sendNotifications();//重新发送通知
                    } else {//连接
                        manager.connectAll();//同步连接上所有有vote资格的sid
                    }

                    /*
                     * Exponential backoff
                     */
                    int tmpTimeOut = notTimeout*2;
                    notTimeout = (tmpTimeOut < maxNotificationInterval?
                            tmpTimeOut : maxNotificationInterval);
                    LOG.info("Notification time out: " + notTimeout);
                }
                else if(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 send messages out
                        if (n.electionEpoch > logicalclock) { //如果接收的epoch比自己的高
                            logicalclock = n.electionEpoch;
                            recvset.clear();//之前接收到的投票作废，必须同一个epoch才行
                            if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                    getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {//如果接收的Notification可以替代自己的提议
                                updateProposal(n.leader, n.zxid, n.peerEpoch);//那么就更新自己的提议
                            } else {//否则还原为初始化提议
                                updateProposal(getInitId(),
                                        getInitLastLoggedZxid(),
                                        getPeerEpoch());
                            }
                            sendNotifications();//给各server发送通知
                        } else if (n.electionEpoch < logicalclock) {//如果接受的epoch比自己的低,就不用管
                            if(LOG.isDebugEnabled()){
                                LOG.debug("Notification election 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)) {//如果epoch一样，并且收到的投票 优于 自己的投票
                            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));//更新收到的票的集合

                        if (termPredicate(recvset,
                                new Vote(proposedLeader, proposedZxid,
                                        logicalclock, proposedEpoch))) {//如果集群验证器验证通过当前机器是"leader"(通常是过半机器投票给当前机器)

                            // Verify if there is any change in the proposed leader
                            while((n = recvqueue.poll(finalizeWait,
                                    TimeUnit.MILLISECONDS)) != null){//接受队列还有Notification
                                if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                        proposedLeader, proposedZxid, proposedEpoch)){
                                    recvqueue.put(n);//如果接收到的投票优于当前自己的投票(当前集群过半投票的机器)
                                    break;
                                }
                            }

                            /*
                             * This predicate is true once we don't read any new
                             * relevant message from the reception queue
                             */
                            if (n == null) {//此时当前投票优于所有接受队列中的通知
                                self.setPeerState((proposedLeader == self.getId()) ?
                                        ServerState.LEADING: learningState());//设置serverState为leading

                                Vote endVote = new Vote(proposedLeader,
                                                        proposedZxid,
                                                        logicalclock,
                                                        proposedEpoch);
                                leaveInstance(endVote);//最终投票结果
                                return endVote;
                            }
                        }
                        break;
                    case OBSERVING:
                        LOG.debug("Notification from observer: " + n.sid);
                        break;
                    case FOLLOWING:
                    case LEADING://收到的通知对应状态为following或leading
                        /*
                         * Consider all notifications from the same epoch
                         * together.
                         */
                        if(n.electionEpoch == logicalclock){
                            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;
                            }
                        }

                        /*
                         * Before joining an established ensemble, verify
                         * a majority is following the same leader.
                         */
                        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-cluster member " + n.sid);
                }
            }
            return null;
        } finally {//jmx相关处理
            try {
                if(self.jmxLeaderElectionBean != null){
                    MBeanRegistry.getInstance().unregister(
                            self.jmxLeaderElectionBean);
                }
            } catch (Exception e) {
                LOG.warn("Failed to unregister with JMX", e);
            }
            self.jmxLeaderElectionBean = null;
        }
    }

这里有两张图描述这个过程
https://mozillazg.github.io/static/images/zookeeper/elect-leader.png
http://7xjtfr.com1.z0.glb.clouddn.com/FLE.png
倾向于后者，如下

lookForLeader流程图

思考

updateProposal和getVote函数什么关系

前者根据参数更新提议即(proposedLeader, proposedZxid, proposedEpoch)
后者是根据提议(proposedLeader, proposedZxid, proposedEpoch)生成投票
可以理解为set和get

getVote()和self.getCurrentVote()的区别是什么

源码中有两种获取vote的形式，这两种的区别是什么
getVote()是临时投票,相当于选leader时的提议，会不断变化的
self.getCurrentVote()是每次选leader时，最后决定下来的投票，一般都是最终的leader

连续多次等待通知都没有等到，等待通知的时长变化如何

FastLeaderElection#lookForLeader

连续等待notify的时长变化

初始值为200ms，也就是说，只要没等到，就再等double的时间，到400ms，800ms，最终不超过maxNotificationInterval也就是1分钟

FastLeaderElection与QuorumCnxManager关系

选leader时消息的收发依赖QuorumCnxManager,它作为server之间的IO管理器，

image.png

Vote为什么要peerEpoch字段

peerEpoch：被推举的Leader的epoch。
前面看代码看peerEpoch都是各种set和get，没有发现哪里比较了，最后发现是
FastLeaderElection#totalOrderPredicate 比较两个vote的大小关系的时候，会先用peerEpoch进行比较

electionEpoch和peerEpoch区别,什么时候会不同?

electionEpoch是选举周期，用于判断是不是同一个选举周期，从0开始累计
peerEpoch是当前周期,用于判断各个server所处的周期，从log中读取currentEpoch

选举leader时，
electionEpoch作为大判断条件，要求大家按最新的electionEpoch作为选举周期
如果electionEpoch一样，那么再根据currentEpoch和zxid，sid等判断哪个server是最“新”的
参考FastLeaderElection#totalOrderPredicate函数

FastLeaderElection.Messenger.WorkerReceiver#run中，初始化时，认为的leader是哪一个

获取初始投票

这里如果是null可是会npe的

ans：值是在这里设置的
QuorumPeer#startLeaderElection，
也就是提前调用了

currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());

初始化时当前投票是投给自己的

两个vote的全序大小比较规则总结

依次根据peerEpoch，zxid,，sid来
peerEpoch代表所处周期，越大则投票越新
peerEpoch相同时，zxid代表一个周期中的事务记录，越大则投票越新
peerEpoch，zxid均相同时，sid大的赢（两个投票一样新，只是为了决定leader需要有大小关系）

见totalOrderPredicate函数

选举投票leader的验证问题

如果消息发送方state是looking，则termPredicate看是否过半即可
如果消息发送方state是following或者leading，则ooePredicate看是否过半，且leader机器发出ack知道了自己是leader即可

集群中是否所有机器是否都是网络互通

这里集群不是所有列表，而是所有leading和following的机器
不是网络互通的，比如

三台机器ABC，AB网络不通
但是A，B，C投票都给C
C收到三张票，过半，自己成为leader
B知道C得到了两张票，分别是BC投给C（B不知道A投给了C），也过半，自己成为follower
同理，A也成为follower

是否会出现looking机器和leader机器网络不通，但收了过半的leader投票，因此认定了leader的合理性

情景：

1.假设5台机器ABCDE，ABCD已经形成集群，以D为leader
2.这时E以looking状态进来，收到了ABC以following状态的投票，这时就过半了
3.E会不会把D当成leader

这就是checkLeader函数的意义
里面会有检查

        if(leader != self.getId()){// 自己不为leader
            if(votes.get(leader) == null) predicate = false;// 投票记录中，没有来自leader的投票
            else if(votes.get(leader).getState() != ServerState.LEADING) predicate = false;//leader不知道自己是leader

如果网络不通，那么就会votes.get(leader) == null，因此E不会把D当成leader

加入一个已经有的集群，走的什么流程

在上面checkLeader意义处也带过了，就是收了一堆following和leader机器的回复，然后进行过半验证以及leader验证即可

竞选leader是"广播"吗？

选举leader不是广播，后续一致性同步才是广播。
这里就是所有server互相通信完成的

leader选举在server中启动步骤的位置

leader选举在server中选举的位置

问题

FastLeaderElection#lookForLeader

里面，处于LOOKING时，没有收到消息时的源码

                    if(manager.haveDelivered()){//如果已经发送过
                        sendNotifications();//重新发送通知
                    } else {//连接
                        manager.connectAll();//同步连接上所有有vote资格的sid
                    }

为什么有不同的逻辑

如果没有连接上，那么也有可能
QuorumCnxManager#queueSendMap还没有put操作
那么调用QuorumCnxManager#connectAll也没用啊

    public void connectAll(){//把所有需要发送消息的机器sid都连接上
        long sid;
        for(Enumeration<Long> en = queueSendMap.keys();//连接所有queueSendMap记录的sid
            en.hasMoreElements();){
            sid = en.nextElement();
            connectOne(sid);
        }      
    }

里面的queueSendMap也没有放入合适的值

代码难度

难理解，尤其是要从各种异常的情况去考虑，如果仅仅看源码，不知道为什么要这样写
比如之前分析checkLeader函数的意义

refer

http://www.cnblogs.com/leesf456/p/6107600.html
http://www.cnblogs.com/leesf456/p/6508185.html
http://codemacro.com/2014/10/19/zk-fastleaderelection/
http://blog.xiaohansong.com/2016/08/25/zab/
http://shift-alt-ctrl.iteye.com/blog/1846562 （多图）
https://my.oschina.net/pingpangkuangmo/blog/778927
《paoxs到zk》 7.6.3节