zookeeper选举流程源码分析

这篇具有很好参考价值的文章主要介绍了zookeeper选举流程源码分析。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

zookeeper选举流程源码分析

选举的代码主要是在QuorumPeer.java这个类中。

它有一个内部枚举类,用来表示当前节点的状态。

   public enum ServerState {
        LOOKING, FOLLOWING, LEADING, OBSERVING;
    }

LOOKING: 当前节点在选举过程中

FOLLOWING:当前节点是从节点

LEADING: 当前节点是主节点

OBSERVING: 当前节点是观察者状态,这种状态的节点不参与选举的投票。

QuorumPeer有个run方法,就是用来根据当前节点不同的状态,进行不同的处理。

下面看下这段代码主要的框架

   @Override
    public void run() {
        updateThreadName();

        LOG.debug("Starting quorum peer");
 				// 这里是注册jmx消息,不用关注
 				//下面就是选举的框架代码了

        try {
      	//running 表示当前节点的状态,只要在运行过程中,就会一直根据当前节点的状态进行不同的处理
            while (running) {
              //getPeerState()用来获取当前节点的状态,就是上面提到的枚举类。
              //下面就会根据不同的状态进行不同的处理
                switch (getPeerState()) {
                case LOOKING:
                    LOG.info("LOOKING");
										......
                     //选举就是调用下面的这行代码来完成的。
                      //后面我们也就单独就这个代码来进行分析
                            setCurrentVote(makeLEStrategy().lookForLeader());
										......
                    break;
                case OBSERVING:
                  	......
                    //按照观察者的逻辑进行处理  
                    break;
                case FOLLOWING:
       							......
                    //按照从节点的逻辑进行处理
                    break;
                case LEADING:
										......
                    //按照主节点的逻辑进行处理
                    break;
                }
                start_fle = Time.currentElapsedTime();
            }
        } finally {
					......
        }
    }

上面代码的逻辑还是比较清楚的,就是一直在这几种状态之间处理。

每种状态的处理逻辑基本都是如下

                    try {
                      //处理业务逻辑,正常情况下,会一直在这里。
                      //除非当前的状态逻辑已经处理完毕,如LOOKING,或者抛出了异常,这时就需要重置状态
                    } catch (Exception e) {
                        LOG.warn("Unexpected exception",e);
                    } finally {
                     	//重置状态
                      updateServerState();
                    }

下面我们看看上面选举的这行代码 setCurrentVote(makeLEStrategy().lookForLeader());。这行代码会调用具体执行选举的类执行具体的选举操作,并返回对应的投票信息,并设置成当前的投票信息。

默认的选举的是FastLeaderElection,对应的选举逻辑就在lookForLeader方法中。下面我们就直接去看看FastLeaderElectionlookForLeader方法吧。

选举的主要逻辑就是告诉其他节点。我是谁,我选谁做为主节点。

public Vote lookForLeader() throws InterruptedException {
				......
        try {
  					//recvset用来保存投票信息,
          //key表示选民身份,也就是这个票是谁投的(注意:每个节点只会有一个有效的投票,后面的投票会覆盖掉之前的投票)
          //value用来表示具体投票的内容
            HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();

            HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();
						
            int notTimeout = finalizeWait;

            synchronized(this){
              	//每次投票前,会先更新这个logicalclock逻辑时钟,这个用来表示当前是第几次选举了,对比投票信息的时候会用到,这个很关键
                logicalclock.incrementAndGet();
              //首先给自己投一票
                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
            }

            LOG.info("New election. My id =  " + self.getId() +
                    ", proposed zxid=0x" + Long.toHexString(proposedZxid));
          	//发送投票信息
            sendNotifications();

在上面的代码,首先会把logicalclock+1,表示当前是启动后的第几轮选取,这个参数是保存在内存中的,也就是每次启动都会从0开始。

那会不会出现节点之间logicalclock不同的情况呢,这个情况是有可能会出现的。不过后面选举过程中,相互发送消息也就会发送logicalclock,会和自己的logicalclock比较,进行修正。

在开始选举的时候,首先会给自己投一票。

会调用sendNotifications方法将投票者(自己)的信息和投票信息发出去。

会发送这些信息:

  • proposedLeader : 选举的主节点
  • proposedZxid: 选举的节点zxid,这个字段是long类型,前面32 bit表示epoch,后面32bit表示事务id
  • logicalclock:投票者的逻辑时钟
  • QuorumPeer.ServerState.LOOKING:投票者的状态(投票的状态肯定是looking)
  • sid: 投票者的id
  • proposedEpoch :选举节点的epoch,也就是proposedZxid的前面32 bit

下面看看具体的选举代码

					//如果当前节点一直是looking,且服务没有停止,就会一直进行选举流程
            while ((self.getPeerState() == ServerState.LOOKING) &&
                    (!stop)){
								// 获取其他节点发送过来的消息
                Notification n = recvqueue.poll(notTimeout,
                        TimeUnit.MILLISECONDS);

      					//如果没有收到消息,就去检查下和其他节点的连接是否正常,尽力使消息能发送。
                if(n == null){
     						......
                } 
                // 验证收到消息的节点和它选举的主节点是否有效
                else if (validVoter(n.sid) && validVoter(n.leader)) {
     								//这里就会根据收到消息的节点状态进行分别进行处理
     								// 比如自己是后加入进来的,这时就已经有了leader节点,对应的也就有follow节点
     								// 也有可能大家都刚启动,或者主节点挂掉了,这时大家都会又会是looking状态
                    switch (n.state) {
                    // 如果对方节点是投票状态
                    case LOOKING:
                        // If notification > current, replace and send messages out
                        //首先比较logicalclock,如果对方的logicalclock比自己的大,就修正自己的`logicalclock`,同时清空自己的票箱,重新计票
                        if (n.electionEpoch > logicalclock.get()) {
                            logicalclock.set(n.electionEpoch);
                            recvset.clear();
                            // 这里会比较票的信息,如果对方选的leader节点的比自己大,就推举对方选的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();
                            // 如果自己的logicalclock 比对方的大,直接忽略对方的票
                        } else if (n.electionEpoch < logicalclock.get()) {
														......
                            break;
                             // 如果logicalclock相等,那就直接比较自己当前选出来的leader和对方选出来的leader进行比较,如果自己的大,就不做处理,如果对方的大,就更新自己的票,重新投票
                        } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                proposedLeader, proposedZxid, proposedEpoch)) {
                            updateProposal(n.leader, n.zxid, n.peerEpoch);
                            sendNotifications();
                        }

												//在这里将对方的票扔进投票箱
                        // don't care about the version if it's in LOOKING state
                        recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
                        
                        //这里就对投票进行统计了,如果过半,就要设置leader了,不过在这之前,会再等一个时间,看看其他节点是否有选出更适合的leader。
                        //如果没有,那就设置对方节点选出来的leader为主节点,对比下leader是不是自己,如果是自己,就将自己的状态修改为leader,否则就修改成follow。同时保存当前leader信息
                       

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

                            /*
                             * 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());
                                Vote endVote = new Vote(proposedLeader,
                                        proposedZxid, logicalclock.get(), 
                                        proposedEpoch);
                                leaveInstance(endVote);
                                return endVote;
                            }
                        }
                        break;
                    case OBSERVING:
                        //这种状态的节点是不参与投票的,所以对它的发送的投票信息进行忽略。
                        break;
                    case FOLLOWING:
                    case LEADING:
//如果对方是following或者leading,说明当前已经有主节点了,在这里就直接统计票数信息,并验证根据票数信息统计出来的leader节点和回应自己消息的自称leader节点 是不是同一个,如果是同一个,说明信息是吻合的,就会去设置自己的节点状态。需要注意的是,投票信息不但会发送给其他节点,也会给自己发送。所以这里会判断对方节点是否是当前节点。
													......
                        break;
                    default:
										......
                    }
                } else {
								......
                }
            }
            return null;
        } finally {
				......
        }
    }

比较节点大小也比较简单。

((newEpoch > curEpoch) ||
                ((newEpoch == curEpoch) &&
                ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))));

首先比较epoch,其次比较zxid,最后比较myid。

myid就是我们在zookeeper每个节点中设置myid文件中对应的值。

zxid是两部分,前32bit epoch,后32 bit 事务序号。在一个节点成为leader节点后,首先会将epoch的值+1,同时将事务序号设置成0。zxid是持久化写入文件的,所以重启也不会丢失。

logicalclock在内存中,所以每次启动都会从0开始。

给其他节点发送投票消息的时候,也会给自己发送,其他节点是通过网络发送,给自己是直接放到接收投票信息的队列。文章来源地址https://www.toymoban.com/news/detail-668048.html

到了这里,关于zookeeper选举流程源码分析的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Apache Zookeeper架构和选举机制

    ZooKeeper是一个开源的分布式协调服务,旨在解决分布式系统中的一致性、配置管理、领导者选举等问题。它由Apache软件基金会维护,是Hadoop生态系统的一部分,被广泛用于构建高可用、可靠和具有一致性的分布式应用程序和服务。 ZooKeeper提供了一个层次化的命名空间,类似于

    2024年02月11日
    浏览(32)
  • zookeeper源码(03)启动流程

    本文将从启动类开始详细分析zookeeper的启动流程: 加载配置的过程 集群启动过程 单机版启动过程 org.apache.zookeeper.server.quorum.QuorumPeerMain类。 用于启动zookeeper服务,第一个参数用来指定配置文件,配置文件properties格式,例如以下配置参数: dataDir - 数据存储目录 dataLogDir - t

    2024年02月08日
    浏览(36)
  • ZooKeeper 选举的过半机制防止脑裂

    结论: Zookeeper采用过半选举机制,防止了脑裂。 原因: 如果有5台节点,leader联系不上了,其他4个节点由于超过半数,所以又选出了一个leader,当失联的leader恢复网络时,发现集群中已经有了leader,会把自己降为flower,防止出现两个leader。 和NameNode不同的是,zookeeper是自己

    2024年02月14日
    浏览(30)
  • ZooKeeper的应用场景(集群管理、Master选举)

    随着分布式系统规模的日益扩大,集群中的机器规模也随之变大,因此,如何更好地进行集群管理也显得越来越重要了。 所谓集群管理,包括集群监控与集群控制两大块,前者侧重对集群运行时状态的收集,后者则是对集群进行操作与控制。在日常开发和运维过程中,我们经

    2024年02月12日
    浏览(24)
  • Redis——哨兵模式与Zookeeper选举的异同点

    当我们使用主从复制出现的问题:手动故障转移:写能力和存储能力受限:主从复制 -master 宕机故障处理。 主从切换技术的方法是:当主服务器宕机后,需要手动把一台从服务器切换为主服务器,这就需要人工干预,费事费力,还会造成一段时间内服务不可用。这不是一种推

    2024年02月06日
    浏览(25)
  • 面试官:说一说Zookeeper中Leader选举机制

    哈喽!大家好,我是小奇,一位不靠谱的程序员 小奇打算以轻松幽默的对话方式来分享一些技术,如果你觉得通过小奇的文章学到了东西,那就给小奇一个赞吧 文章持续更新,可以微信搜索【小奇JAVA面试】第一时间阅读,回复【资料】更有我为大家准备的福利哟! 今天又是

    2023年04月08日
    浏览(33)
  • zookeeper源码(08)请求处理及数据读写流程

    用于接收客户端连接、管理客户端session、处理客户端请求。 代表一个客户端连接对象: 从网络读写数据 数据编解码 将请求转发给上层组件或者从上层组件接收响应 管理连接状态,比如:enableRecv、sessionTimeout、stale、invalid等 保存当前的packetsReceived、packetsSent、lastCxid、last

    2024年02月19日
    浏览(26)
  • Zookeeper快速入门(Zookeeper概述、安装、集群安装、选举机制、命令行操作、节点类型、监听器原理)

    1.1 概述 Zookeeper是一个开源的分布式的,为分布式框架提供协调服务的Apache项目。 1、Zookeeper工作机制 Zookeeper从设置模式角度来理解:是一个基于观察者模式设计的分布式服务管理框架,它负责储存和管理大家都关心的数据,然后接受观察者的注册,一旦这些数据的状态发生

    2024年03月28日
    浏览(40)
  • Kafka 和 Zookeeper 的 Leader 和 Follower 区别(选举&数据同步)

    》Client与Server通过NIO通信 》全局串行化所有的写操作 》保证同一客户端的指令被FIFO执行 》保证消息通知的FIFO (2)kafka 不同,只有leader 负责读写,follower只负责备份,如果leader宕机的话,Kafaka动态维护了一个同步状态的副本的集合(a set of in-sync replicas),简称ISR,ISR中有f+1个节

    2024年04月10日
    浏览(54)
  • 标题6: ZooKeeper源码分析

    作者:禅与计算机程序设计艺术 ZooKeeper是一个开源分布式协调服务框架,其设计目标是将那些复杂且容易出错的分布式一致性服务封装成简单易用、高性能的客户端接口。它是Google的Chubby、Google文件系统GFS和Apache Hadoop的子项目之一。2010年,ZooKeeper正式成为Apache顶级项目。在

    2024年02月06日
    浏览(21)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包