千家信息网

如何对FastLeaderElection进行源码解析

发表于:2025-01-31 作者:千家信息网编辑
千家信息网最后更新 2025年01月31日,这篇文章将为大家详细讲解有关如何对FastLeaderElection进行源码解析,文章内容质量较高,因此小编分享给大家做个参考,希望大家阅读完这篇文章后对相关知识有一定的了解。zookeeper作为
千家信息网最后更新 2025年01月31日如何对FastLeaderElection进行源码解析

这篇文章将为大家详细讲解有关如何对FastLeaderElection进行源码解析,文章内容质量较高,因此小编分享给大家做个参考,希望大家阅读完这篇文章后对相关知识有一定的了解。

zookeeper作为常用的分布式协调器、注册中心,在初始化启动、leader宕机、或者与leader心跳超时等情况下,为保证集群可用性会进行重新选主。选主的默认算法是FastLeaderElection,本文会对FastLeaderElection进行解析。

选主入口

在QuorumPeer.run()方法中,进行主循环while(running)。如果当前状态为LOOKING,则进入重新选主的方法lookForLeader。

//主循环while (running){    ...    switch (getPeerState()) {        case LOOKING:            ...            //调用makeLEStrategy获取选主策略,默认为FastLeaderElection            //进入lookForLeader方法,开启重新选主。            //把lookForLeader返回的结果调用setCurrentVote缓存起来。            setCurrentVote(makeLEStrategy().lookForLeader());            ...            break;        case OBSERVING:            ...            break;        case FOLLOWING:            ...            break;        case LEADING:            break;    }}

选主流程

进入FastLeaderElection#lookForLeader后,代码大致逻辑如下

1.leader选举周期版本号+1,发起第一次投票,默认选自己。

2.开启循环,接收其他zk节点发来的投票通知

2.1接收不到其他节点的投票时,指数级延长接收等待时间并重新执行步骤2的循环

2.2接收到其他节点投票后,判断投票通知来源节点的状态

2.2.1投票来源节点为LOOKING状态,比较投票通知和本地投票信息缓存。比较投票周期版本号、决议zxid、zk节点的序号,判断哪个更有效。

2.2.1.2比较后重新计算本机当前的投票信息,包括推选的leader id、推选leader的决议zxid、leader选举周期版本号,并发送给集群其他节点

2.2.1.3判断本机最新的投票信息,在本机接收到的所有投票通知缓存中的数量,判断是否超过集群总数的一半

2.2.1.3.1未超过集群总数的一半,重新执行步骤2的循环

2.2.1.3.2超过集群总数的一半,尝试循环拉取投票队列中剩余的其他票。

2.2.1.3.2.1如果拉取到了其他投票,过滤掉无效的票,重新执行步骤2的循环。

2.2.1.3.2.2如果没有拉取到其他选票,设置当前节点的状态,LEADING、 FOLLOWING或者OBSERVING,清除缓存,返回最终leader的结果。

2.2.2投票来源节点为OBSERVING,此状态的节点不能投票,丢弃此票,继续执行步骤2的循环。

2.2.3投票来源节点为FOLLOWING,说明在集群内部已经选出主节点了,此时放弃自己的投票,查询outofelection中的其他节点的投票缓存,直接确定leader。设置当前节点状态为FOLLOWING或者OBSERVING,返回选票结果。

2.2.4投票来源节点为LEADING,说明在集群内部已经选出主节点了,此时放弃自己的投票,查询outofelection中的其他节点的投票缓存,直接确定leader。设置当前节点状态为FOLLOWING或者OBSERVING,返回选票结果。

    /**     * 开始新一轮leader选举。每当我们的服务节点状态变更为LOOKING时,调用此方法,然后向所有其他对等方发送通知     */    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;        }        self.start_fle = Time.currentElapsedTime();        try {            /**             * 当前领导人选举的选票存储在recvset这个map中。换句话说,选票Vote在recvset中             * 必须满足条件v.electionEpoch == logicalclock(判定这个选票,是否属于这次选举周期)。             * 当前选举参与节点使用recvset来推断是否大多选举参与节点投了赞成票。             */            Map recvset = new HashMap();            /**             * 历次leader选举的票数以及本次leader选举的票数均会存储在outofelection中。             * 注意,处于LOOKING状态的通知不会存储在outofelection中。             *             * 如果当前节点参与选举时,集群其他节点已经选举出了leader,则根据这个map中的数据直接跟随集群leader             */            Map outofelection = new HashMap();            int notTimeout = minNotificationInterval;            synchronized (this) {                //选举的逻辑周期+1,表示发起新一届选举                logicalclock.incrementAndGet();                //更新投票信息,第一个参数表示选谁(默认选自己),第二个参数表示当前本地日志中最新的决议zxid,第三个参数表示新的leader周期的epoch编号                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());            }            LOG.info(                "New election. My id = {}, proposed zxid=0x{}",                self.getId(),                Long.toHexString(proposedZxid));            //发送投票信息,广播给其他节点            sendNotifications();            SyncedLearnerTracker voteSet = null;            /*             * 在循环中,我们和其他zk节点交换通知,直到找到一个leader             */            while ((self.getPeerState() == ServerState.LOOKING) && (!stop)) {                //从接收队列中,拉取一个其他节点的投票通知,拉取超时时间为200ms                Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);                //200ms内没有拉取到新的投票通知,则发送更多投票通知。                //拉取到了则否则处理新投票通知。                if (n == null) {                    if (manager.haveDelivered()) {                        sendNotifications();                    } else {                        manager.connectAll();                    }                    /*                     * 指数级延长等待时间                     */                    int tmpTimeOut = notTimeout * 2;                    notTimeout = Math.min(tmpTimeOut, maxNotificationInterval);                    self.getQuorumVerifier().revalidateVoteset(voteSet, notTimeout != minNotificationInterval);                    if (self.getQuorumVerifier() instanceof QuorumOracleMaj && voteSet != null && voteSet.hasAllQuorums() && notTimeout != minNotificationInterval) {                        setPeerState(proposedLeader, voteSet);                        Vote endVote = new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch);                        leaveInstance(endVote);                        return endVote;                    }                    LOG.info("Notification time out: {} ms", notTimeout);                } else if (validVoter(n.sid) && validVoter(n.leader)) {                    //判断这个投票的来源节点的状态                    switch (n.state) {                        //处于选举中的的状态                    case LOOKING:                        //当前日志中最新的zxid为-1,属于异常情况,直接退出                        if (getInitLastLoggedZxid() == -1) {                            LOG.debug("Ignoring notification as our zxid is -1");                            break;                        }                        //投票通知中的zxid为-1,也属于异常情况,直接退出                        if (n.zxid == -1) {                            LOG.debug("Ignoring notification from member with -1 zxid {}", n.sid);                            break;                        }                        //判断投票通知中的leader选举周期版本号,是否大于本地选举周期中的版本号                        if (n.electionEpoch > logicalclock.get()) {                            //本地leader选举周期版本号不是最新的,更新成更新的leader选举周期版本号                            logicalclock.set(n.electionEpoch);                            //清除空旧的选票map                            recvset.clear();                            //调用核心方法totalOrderPredicate                            // 投票通知中的信息和本地初始化信息对比,判断投票是否有效                            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()) {                            //接收的投票通知中的leader选举周期版本号比本地的旧,则丢弃这个投票通知                                LOG.debug(                                    "Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x{}, logicalclock=0x{}",                                    Long.toHexString(n.electionEpoch),                                    Long.toHexString(logicalclock.get()));                            break;                        } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) {                            //还是调用totalOrderPredicate方法                            // 用投票通知中的信息,和本地当前缓存中最新的投票信息进行比对,如果投票中的比较新,则进入这里                            updateProposal(n.leader, n.zxid, n.peerEpoch);                            sendNotifications();                        }                        LOG.debug(                            "Adding vote: from={}, proposed leader={}, proposed zxid=0x{}, proposed election epoch=0x{}",                            n.sid,                            n.leader,                            Long.toHexString(n.zxid),                            Long.toHexString(n.electionEpoch));                        //在选票map中存储选票                        recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));                        //proposedLeader、proposedZxid、proposedEpoch三个参数,表示当前zk服务根据已经接收到的投票,确定出的本机投选的leader信息                        //调用getVoteTracker方法,对本机投选的leader的选票进行统计                        voteSet = getVoteTracker(recvset, new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch));                        //本机投选的leader是否获取了半数以上的票                        if (voteSet.hasAllQuorums()) {                            // 拉取投票接收队列中的剩余选票                            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) {                                //根据投票结果,设置当前节点的状态,LEADING、 FOLLOWING或者OBSERVING                                setPeerState(proposedLeader, voteSet);                                //创建最终选举结果的选票对象                                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:                        //收到的投票通知中,投票来源的机器是FOLLOWING,说明在集群内部已经选出主节点了                        //此时放弃自己的投票,查询outofelection中的投票,直接确定leader                        Vote resultFN = receivedFollowingNotification(recvset, outofelection, voteSet, n);                        if (resultFN == null) {                            break;                        } else {                            return resultFN;                        }                    case LEADING:                        //收到的投票通知中,投票来源的机器是LEADING,说明在集群内部已经选出主节点了                        //此时放弃自己的投票,查询outofelection中的投票,直接确定leader                        Vote resultLN = receivedLeadingNotification(recvset, outofelection, voteSet, n);                        if (resultLN == null) {                            break;                        } else {                            return resultLN;                        }                    default:                        LOG.warn("Notification state unrecognized: {} (n.state), {}(n.sid)", n.state, n.sid);                        break;                    }                } else {                    if (!validVoter(n.leader)) {                        LOG.warn("Ignoring notification for non-cluster member sid {} from sid {}", n.leader, n.sid);                    }                    if (!validVoter(n.sid)) {                        LOG.warn("Ignoring notification for sid {} from non-quorum member sid {}", n.leader, n.sid);                    }                }            }            return null;        } finally {            try {                if (self.jmxLeaderElectionBean != null) {                    MBeanRegistry.getInstance().unregister(self.jmxLeaderElectionBean);                }            } catch (Exception e) {                LOG.warn("Failed to unregister with JMX", e);            }            self.jmxLeaderElectionBean = null;            LOG.debug("Number of connection processing threads: {}", manager.getConnectionThreadCount());        }    }

核心投票有效性抉择逻辑

选LEADER主流程中,多次调用FastLeaderElection#totalOrderPredicate进行了投票有效性的比较,该逻辑是选主有效性抉择的核心逻辑。优先比较leader选举周期版本号epoch,相等时比较决议zxid,都相等时直接使用leader机器id大的。

 /**     * 检查接收到的投票通知,是否为有效投票     * @param newId 新投票中的leader节点id     * @param newZxid 新投票中的决议id     * @param newEpoch 新投票中的leader选举周期版本号     * @param curId 当前本机投票中的leader节点id     * @param curZxid 当前本机投票中的决议id     * @param curEpoch 当前本机投票中的leader选举周期版本号     * @return 新投票更有效时返回true,否则false     */    protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {        LOG.debug(            "id: {}, proposed id: {}, zxid: 0x{}, proposed zxid: 0x{}",            newId,            curId,            Long.toHexString(newZxid),            Long.toHexString(curZxid));        if (self.getQuorumVerifier().getWeight(newId) == 0) {            return false;        }        /*         *如果以下三种情况之一成立,则返回true:         *1-投票中的leader选举周期版本号大于本地的         *2-leader选举版本号相同,但投票中的决议号zxid更高         *3-leader选举版本号和决议号zxid都相同,则比较zk机器本身的机器编号,判断投票中的机器编号是否大于当前机器编号         */        return ((newEpoch > curEpoch)                || ((newEpoch == curEpoch)                    && ((newZxid > curZxid)                        || ((newZxid == curZxid)                            && (newId > curId)))));    }

关于如何对FastLeaderElection进行源码解析就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看到。

0