美文网首页
品味Zookeeper之选举及数据一致性

品味Zookeeper之选举及数据一致性

作者: dandan的微笑 | 来源:发表于2018-03-09 21:36 被阅读0次

    品味Zookeeper之选举及数据一致性

    本文思维导图

    image

    前言

    为了高可用和数据安全起见,zk集群一般都是由几个节点构成(由n/2+1,投票机制决定,肯定是奇数个节点)。多节点证明它们之间肯定会有数据的通信,同时,为了能够使zk集群对外是透明的,一个整体对外提供服务,那么客户端访问zk服务器的数据肯定是要数据同步,也即数据一致性

    zk集群是Leader/Follower模式来保证数据同步的。整个集群同一时刻只能有一个Leader,其他都是Follower或Observer。Leader是通过选举选出来的,这里涉及到ZAB协议(原子消息广播协议)。

    1.ZAB协议

    1.1 概念理解

    为了更好理解下文,先说ZAB协议,它是选举过程和数据写入过程的基石。ZAB的核心是定义会改变zk服务器数据状态的事务请求的处理方式。

    ZAB的理解:所有事务请求是由一个全局唯一的服务器来协调处理,这个的服务器就是Leader服务器,
    其它服务器都是Follower服务器或Observer服务器。Leader服务器负责将一个客户端的请求转换成那个一个事务Proposalͧ(提议),将该Proposal分发给集群中所有的Follower服务器。然后Leader服务器需要等待所有Follower服务器的应答,当Leader服务器收到超过半数的Follower服务器进行了明确的应答后,Leader会再次向所有的Follower服务器分发Commit消息,要求其将前一个Proposal进行提交。

    注意事务提议这个词,就类似 人大代表大会提议 ,提议就代表会有应答,之间有通信。因此在zk的ZAB协议为了可靠性和可用性,会有投票应答等操作来保证整个zk集群的正常运行。

    总的来说就是,涉及到客户端对zk集群数据改变的行为都先由Leader统一响应,然后再把请求转换为事务转发给其他所有的Follower,Follower应答并处理事务,最后再反馈。如果客户端只是读请求,那么zk集群所有的节点都可以响应这个请求。

    1.2 ZAB协议三个阶段

    • 1.发现(选举Leader过程)
    • 2.同步(选出Leader后,Follower和Observer需进行数据同步)
    • 3.广播(同步之后,集群对外工作响应请求,并进行消息广播,实现数据在集群节点的副本存储)

    下面会逐点分析,但是在这之前先来了解了解zookeeper服务器的知识吧。

    2.Zookeeper服务器

    2.1 zk服务器角色

    • Leader
    • 事务请求的唯一调度和处理者,保证集群事务处理的顺序序性
    • 集群内部各服务器的调度者
    • Follower
    • 处理客户端非事务请求,转发事务请求给Leader服务器
    • 参与事务请求Proposal的投票
    • 参与Leader的选举投票
    • Observer
    • 处理客户端非事务请求,转发事务请求给Leader服务器
    • 不参加任何形式的投票,包括选举和事务投票(超过半数确认)
    • Observer的存在是为了提高zk集群对外提供读性能的能力

    整个zk集群的角色作用如下图:

    image

    2.2 zk服务器状态

    • LOOKING
    • 寻找Leader状态
    • 当服务器处于这种状态时,表示当前没有Leader,需要进入选举流程
    • FOLLOWING
    • 从机状态,表明当前服务器角色是Follower
    • OBSERVING
    • 观察者状态,表明当前服务器角色是Observer
    • LEADING
    • 领导者状态,表明当前服务器角色是Leader
    • ServerState 类维护服务器四种状态。


      image

    zk服务器的状态是随着机器的变化而变化的。比如Leader宕机了,服务器状态就变为LOOKING,通过选举后,某机器成为Leader,服务器状态就转换为LEADING。其他情况类似。

    2.3 zk服务器通信

    集群嘛,节点之间肯定是要通信的。zokeeper通信有两个特点:

    • 1.使用的通信协议是TCP协议。在集群中到底是怎么连接的呢?还记得在配置zookeeper时要创建一个data目录并在其他创建一个myid文件并写入唯一的数字吗?zk服务器的TCP连接方向就是依赖这个myid文件里面的数字大小排列。数小的向数大的发起TCP连接。比如有3个节点,myid文件内容分别为1,2,3。zk集群的tcp连接顺序是1向2发起TCP连接,2向3发起TCP连接。如果有n个节点,那么tcp连接顺序也以此类推。这样整个zk集群就会连接起来。

    • 2.zk服务器是多端口的。例如配置如下:

        tickTime=2000
        dataDir=/home/liangjf/app/zookeeper/data
        dataLogDir=/home/liangjf/app/zookeeper/log
        clientPort=2181
        initLimit=5
        syncLimit=2
        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个端口是通信和数据同步端口,默认是2888

    • 第2个端口是投票端口,默认是3888

    3.选举机制

    3.1 选举算法

    从zookeeper开始发布以来,选举的算法也慢慢优化。现在为了可靠性和高可用,从3.4.0版本开始zookeeper只支持基于TcpFastLeaderElection选举协议。

    • LeaderElection
    • Udp协议
    • AuthFastLeaderElection
    • udp
    • FastLeaderElection
    • Udp
    • Tcp

    FastLeaderElection选举协议使用TCP实现Leader投票选举算法。它使用了类对象quorumcnxmanager管理连接。该算法是基于推送的,可以通过调节参数来改变选举的过程。第一,finalizewait决定等到决定Leader的时间。这是Leader选举算法的一部分。
    final static int finalizeWait = 200;(选举Leader过程的进程时间)
    final static int maxNotificationInterval = 60000;(通知检查选中Leader的时间间隔)
    final static int IGNOREVALUE = -1;
    这里先不详细分析,下面3.5 选举算法源码分析及举栗子才分析。

    3.2 何时触发选举

    选举Leader不是随时选举的,毕竟选举有产生大量的通信,造成网络IO的消耗。因此下面情况才会出现选举:

    • 集群启动
    • 服务器处于寻找Leader状态
    • 当服务器处于LOOKING状态时,表示当前没有Leader,需要进入选举流程
    • 崩溃恢复
    • Leader宕机
    • 网络原因导致过半节点与Leader心跳中断

    3.3 如何成为Leader

    • 数据新旧程度
    • 只有拥有最新数据的节点才能有机会成为Leader
    • 通过zxid的大小来表示数据的新,zxid越大代表数据越新
    • myid
    • 集群启动时,会在data目录下配置myid文件,里面的数字代表当前zk服务器节点的编号
    • 当zk服务器节点数据一样新时, myid中数字越大的就会被选举成ОLeader
    • 当集群中已经有Leader时,新加入的节点不会影响原来的集群
    • 投票数量
    • 只有得到集群中多半的投票,才能成为Leader
    • 多半即:n/2+1,其中n为集群中的节点数量

    3.4 重要的zxid

    由3.3知道zxid是判断能否成为Leader的条件之一,它代表服务器的数据版本的新旧程度。

    zxid由两部分构成:主进程周期epoch和事务单调递增的计数器。zxid是一个64位的数,高32位代表主进程周期epoch,低32位代表事务单调递增的计数器

    主进程周期epoch也叫epoch,是选举的轮次,每选举一次就递增1。事务单调递增的计数器在每次选举完成之后就会从0开始。

    如果是比较数据新旧的话,直接比较就可以了。因为如果是主进程周期越大,即高32位越大,那么低32位就不用再看了。如果主进程周期一致,低32位越大,整个zxid就越大。所以直接比较整个64位就可以了,不必高32位于高32位对比,低32位与低32位比较。

    3.5 选举算法源码分析及举栗子

    3.5.1 举栗子

    zookeeper选举有两种情况:

    • 1.集群首次启动
    • 2.集群在工作时Leader宕机

    选主原则如下(在选举时,对比次序是从上往下)

    • 1.New epoch is higher
    • 主周期更大,代所有一切是最新,就成为leader
    • 2.New epoch is the same as current epoch, but new zxid is higher
    • 主周期一致就是在同一轮选票中,zxid越大就成为leader,因为数据更新
    • 3.New epoch is the same as current epoch, new zxid is the same as current zxid, but server id is higher
    • 主周期和zxid一致,就看机器的id(myid),myid越大就成为leader

    同时,在选举的时候是投票方式进行的,除主进程周期外,投票格式为(myid,zxid)。

    第一种情况,比较容易理解,下面以3台机器为例子。

    • 三个zk节点A,B,C,三者开始都没有数据,即Zxid一致,对应的myid为1,2,3。
    • A启动myid为1的节点,zxid为0,此时只有一台服务器无法选举出Leader
    • B启动myid为2的节点,zxid为0,B的zxid与A一样,比较myid,B的myid为2比A为1大,B成Leader
    • C启动myid为3的节点,因为已经有Leader节点,则C直接加入集群,承认B是leader

    第二种情况,已5台机器为例子。

    • 五个节点A,B,C,D,E,B是Leader,其他是Follower,myid分别为1,2,3,4,5,zxid分别为3,4,5,6,6。运行到某个时刻时A,B掉线或宕机,此时剩下C D E。在同一轮选举中,C,D,E分别投自己和交叉投票。
    • 第一次投票,都是投自己。
    • 投票情况为:C:(3,5) D:(4,6) E:(5,6)。
    • 同时也会收到其他机器的投票。
    • 投票情况为:C:(3,5)(4,6)(5,6),D:(4,6)(3,5)(5,6),E:(5,6)(4,6)(3,5)
    • 机器内部会根据选主原则对比投票,变更投票,投票情况为:C:(3,5)(4,6)(5,6)【不变更】。 D:(4,6)(4,6)(5,6)【变更】。E:(5,6)(5,6)(5,6)【变更】
    • 统计票数,C-1票,D-3票,E-5票。因此E成为Leader。

    接下来就是对新Leader节点的检查,数据同步,广播,对外提供服务。

    3.5.1 选举算法源码分析

    选举算法的全部代码在FastLeaderElection类中。其他的lookForLeader函数是选举Leader的入口函数。

    //每一轮选举就会增大一次逻辑时钟,同时更新事务
    synchronized(this){
        logicalclock++;
        updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
    }
    

    //一直循环选举直到找到leader,这里把打印和不相关的都删除了,方便分析。

    while ((self.getPeerState() == ServerState.LOOKING) &&
            (!stop)){
    
        //从通知队列拉取一个投票通知
        Notification n = recvqueue.poll(notTimeout,
                TimeUnit.MILLISECONDS);
    
        if(n == null){
            //看是否选举时通知发送/接收超时
            int tmpTimeOut = notTimeout*2;
            notTimeout = (tmpTimeOut < maxNotificationInterval?
                    tmpTimeOut : maxNotificationInterval);
        }
        else if(self.getVotingView().containsKey(n.sid)) {
            switch (n.state) {
            case LOOKING://只有zk服务器状态为LOOKING时才会进行选举
                // If notification > current, replace and send messages out
                if (n.electionEpoch > logicalclock) {
                    //如果选举时的逻辑时钟大于发送通知来源的机器的逻辑时钟,就把对方的修改为自己的。
                    logicalclock = n.electionEpoch;
                    recvset.clear();
                    //并统计票数,如果能成为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) {
                    //如果通知来演的机器的逻辑时钟比本次我的选举时钟低,直接返回,什么都不做。因为对方没机会成为leader
                    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一样,就看zxid的比较。不过还是会更新事务和回传通知
                    updateProposal(n.leader, n.zxid, n.peerEpoch);
                    sendNotifications();
                }
    
                //把所有接收到的投票信息都放到recvset集合
                recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
    
                //统计谁的投票超过半数,就成为leader
                if (termPredicate(recvset,
                        new Vote(proposedLeader, proposedZxid,
                                logicalclock, proposedEpoch))) {
    
                    //验证一下,被选举的leader是否有变化,就是看符不符合
                    while((n = recvqueue.poll(finalizeWait,
                            TimeUnit.MILLISECONDS)) != null){
                        if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                proposedLeader, proposedZxid, proposedEpoch)){
                            //符合就放进recvqueue集合
                            recvqueue.put(n);
                            break;
                        }
                    }
    
                    //改变选举为leader的机器的状态为LEADING
                    if (n == null) {
                        self.setPeerState((proposedLeader == self.getId()) ?
                                ServerState.LEADING: learningState());
    
                        Vote endVote = new Vote(proposedLeader,
                                proposedZxid, proposedEpoch);
                        leaveInstance(endVote);
                        return endVote;
                    }
                }
                break;
            case FOLLOWING:
            case LEADING:
                //在同一轮选举中,判断所有的通知,并确认自己是leader
                if(n.electionEpoch == logicalclock){
                    recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
                    if(termPredicate(recvset, new Vote(n.leader,
                                    n.zxid, n.electionEpoch, n.peerEpoch, n.state))
                                    && checkLeader(outofelection, n.leader, n.electionEpoch)) {
                        self.setPeerState((n.leader == self.getId()) ?
                                ServerState.LEADING: learningState());
    
                        Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
                        leaveInstance(endVote);
                        return endVote;
                    }
                }
                //在对外提供服务前,先广播一次自己是leader的消息给所有follower,让大家认同我为leader。
                outofelection.put(n.sid, new Vote(n.leader, n.zxid,
                        n.electionEpoch, n.peerEpoch, n.state));
                if (termPredicate(outofelection, new Vote(n.leader,
                        n.zxid, n.electionEpoch, n.peerEpoch, n.state))
                        && checkLeader(outofelection, n.leader, n.electionEpoch)) {
                    synchronized(this){
                        logicalclock = n.electionEpoch;
                        self.setPeerState((n.leader == self.getId()) ?
                                ServerState.LEADING: learningState());
                    }
                    Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
                    leaveInstance(endVote);
                    return endVote;
                }
                break;
            }
        } 
    }
    

    比较重要的子函数有以下这些:

    • 1.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){
            return false;
        }
        //按照这样的顺序比较优先:Epoch > Zxid > myid
        return ((newEpoch > curEpoch) || 
                ((newEpoch == curEpoch) &&
                ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))));
    }
    
    • 2.termPredicate。(最终的计算票数。先把投票放到集合中,然后再统计。集合能去重)

    private boolean termPredicate(
            HashMap<Long, Vote> votes,
            Vote vote) {
    
        HashSet<Long> set = new HashSet<Long>();
        for (Map.Entry<Long,Vote> entry : votes.entrySet()) {
            if (vote.equals(entry.getValue())){
                set.add(entry.getKey());
            }
        }
        return self.getQuorumVerifier().containsQuorum(set);
    }
    
    • 3.Messenger。(构造Messenger的时候创建2条线程WorkerSender和WorkerReceiver用于整个选举的集群投票通信)

    Messenger(QuorumCnxManager manager) {
        this.ws = new WorkerSender(manager);
        Thread t = new Thread(this.ws,
                "WorkerSender[myid=" + self.getId() + "]");
        t.setDaemon(true);
        t.start();
    
        this.wr = new WorkerReceiver(manager);
        t = new Thread(this.wr,
                "WorkerReceiver[myid=" + self.getId() + "]");
        t.setDaemon(true);
        t.start();
    }
    

    其他细节不多说了,主要是sendqueue和recvqueue队列存放待发送投票通知和接收投票通知,WorkerSender和WorkerReceiver两条线程用于投票的通信,QuorumCnxManager manager用于真正和其他机器的tcp连接维护管理,Messenger是整个投票通信的管理者。

    3.数据同步机制

    3.1 同步准备

    完成选举之后,为了数据一致性,需要进行数据同步流程。

    3.1.1 Leader准备
    • Leader告诉其它follower当前最新数据是什么即zxid
    • Leader构建一个NEWLEADER的包,包括当前最大的zxid,发送给所有的follower或者Observer
    • Leader给每个follower创建一个线程LearnerHandler来负责处理每个follower的数据同步请求,同时主线程开始阻塞,等到超过一半的follwer同步完成,同步过程才完成,leader才真正成为leader
    3.1.2 Follower准备
    • 选举完成后,尝试与leader建立同步连接,如果一段时间没有连接上就报连接超时,重新回到选举状态FOLLOWING
    • 向leader发送FOLLOWERINFO包,带上follower自己最大的zxid

    3.2 同步初始化

    同步初始化涉及到三个东西:minCommittedLog、maxCommittedLog、zxid
    – minCommittedLog:最小的事务日志id,即zxid没有被快照存储的日志文件的第一条,每次快照存储
    完,会重新生成一个事务日志文件
    – maxCommittedLog:事务日志中最大的事务,即zxid

    4.数据同步场景

    • 直接差异化同步(DIFF同步)
    • 仅回滚同步TRUNCͨ,即删除多余的事务日志,比如原来的Leader宕机后又重新加入,可能存在它自己写
      入提交但是别的节点还没来得及提交
    • 先回滚再差异化同步(TRUNC+DIFF同步)
    • 全量同步(SNAP同步)

    不同的数据同步算法适用不同的场景。

    5.广播流程

    • 集群选举完成,并且完成数据同步后,开始对外服务,接收读写请求
    • 当leader接收到客户端新的事务请求后,会生成对新的事务proposal,并根据zxid的顺序向所有的
      follower分发事务proposal
    • 当follower收到leader的proposal时,根据接收的先后顺序处理proposal
    • 当Leader收到follower针对某个proposal过半的ack后,则发起事务提交,重新发起一个commit的
      proposal
    • Follower收到commit的proposal后,记录事务提交,并把数据更新到内存数据库
    • 补充说明
    • 由于只有过半的机器给出反馈,则可能存在某时刻某些节点数据不是最新的
    • 如果需要确定读取到的数据是最新的,则可以在读取之前,调用sync方法进行数据同步

    6.小结

    在zookeeper中,除了watcher机制,会话管理,最重要的就是选举了。它是zookeeper集群的核心,也是广泛应用在商业中的前提。洋洋洒洒一大篇,可能存在一些不足,后面更加深入理解再来补充吧。

    相关文章

      网友评论

          本文标题:品味Zookeeper之选举及数据一致性

          本文链接:https://www.haomeiwen.com/subject/xobmfftx.html