CSDN地址：https://blog.csdn.net/Eclipse_2019/article/details/126418783

学习目标

Zookeeper集群中的Leader选举
Zookeeper集群中的数据同步

第1章集群简介

前面第1篇文章讲过Zookeeper是作为分布式环境下的协调器而出生的，既然它是在分布式环境使用的，那它本身也要做集群才能保证高可用性。在前文中也讲过了源码的集群部署和Centos7系统下的集群部署，那么这里我们先通过一张图来大概了解一下集群时的启动和单机版的启动的不同之处：

从这张图上，我们可以看出来，集群启动和单机启动的最大不同就是集群启动的时候会配置选举的相关参数，以及启动选举策略。这其实说白了，就是解决在集群环境下如何保证数据的一致性问题的。

本文将从两个方面来学习Zookeeper的集群：1、Zookeeper集群是如何做数据同步的。2、Zookeeper中leader挂掉了如何进行选举。明确了这两个点，那我们Zookeeper的数据一致性就不存在问题了。

接下来我们先了解一下集群中有哪几种角色。其实我们之前学过redis集群或者mysql集群等知识，大体的也能猜到，既然要做集群，不同的节点肯定有自己的身份。那在zookeeper的集群中，实际上分为三种角色：分别是Leader、Follower、Observer。

1.1 集群角色

1、Leader：是整个zookeeper集群的核心，一个集群中只存在一个Leader节点，它主要的工作任务有两项

事务请求的唯一调度和处理者，保证集群事务处理的顺序性
集群内部各服务器的调度者

2、Follower：主要职责是

处理客户端非事物请求、转发事物请求给leader服务器
参与事物请求Proposal的投票（需要半数以上服务器通过才能通知leader commit数据; Leader发起的提案，要求Follower投票）
参与Leader选举的投票

3、Observer：zookeeper3.3开始引入的一个全新的服务器角色，从字面来理解，该角色充当了观察者的角色。

观察zookeeper集群中的最新状态变化并将这些状态变化同步到observer服务器上。Observer的工作原理与follower角色基本一致，而它和follower角色唯一的不同在于observer不参与任何形式的投票，包括事物请求Proposal的投票和leader选举的投票。简单来说，observer服务器只提供非事物请求服务，通常在于不影响集群事物处理能力的前提下提升集群非事物处理的能力。

接下来再分析一下，在zookeeper集群中，各节点是遵循什么样的协议。

1.2 ZAB协议

1.2.1 概念

全称 Zookeeper Atomic Broadcast（Zookeeper 原子广播协议）。它是专门为Zookeeper设计的一种支持崩溃恢复和原子广播的协议。

ZAB协议包括两种基本的模式：崩溃恢复和消息广播

当整个服务框架在启动过程中，或是当Leader服务器出现网络中断崩溃退出与重启等异常情况时，ZAB就会进入恢复模式并选举产生新的Leader服务器。

当选举产生了新的Leader服务器，同时集群中已经有过半的机器与该Leader服务器完成了状态同步之后，ZAB协议就会退出崩溃恢复模式，进入消息广播模式。

当有新的服务器加入到集群中去，如果此时集群中已经存在一个Leader服务器在负责进行消息广播，那么新加入的服务器会自动进入数据恢复模式，找到Leader服务器，并与其进行数据同步，然后一起参与到消息广播流程中去。

以上其实大致经历了三个步骤：

崩溃恢复：主要就是Leader选举过程
数据同步：Leader服务器与其他服务器进行数据同步
消息广播：Leader服务器将数据发送给其他服务器

1.2.2 结论

ZooKeeper保证的是CP

ZooKeeper不能保证每次服务请求的可用性。（注：在极端环境下，ZooKeeper可能会丢弃一些请求，消费者程序需要重新请求才能获得结果）。所以说，ZooKeeper不能保证服务可用性。
进行Leader选举时集群都是不可用

上面讲了ZAB协议是Zookeeper集群中进行数据同步的基本协议，我们在学习Zookeeper的时候经常会听到ZAB协议和Paxos算法，实际上这是两个不同的东西，Zookeeper是基于ZAB协议去做的数据同步，而ZAB协议底层又是通过Paxos算法去完成Leader选举的，说准确点，应该是FastLeaderElection算法去进行Leader选举的，但FastLeaderElection算法又是典型的Paxos算法，所以我们先来了解下Paxos算法，这样更有助于掌握FastLeaderElection算法。

1.3 Paxos算法

1.3.1 介绍

分布式事务中常见的事务模型有2PC和3PC，无论是2PC提交还是3PC提交都无法彻底解决分布式的一致性问题。这个在之前讲Seata的时候已经介绍过。Google Chubby的作者Mike Burrows说过，世上只有一种一致性算法，那就是Paxos，所有其他一致性算法都是Paxos算法的不完整版，如Chubby， Raft，ZAB，微信的PhxPaxos等。Paxos算法是莱斯利·兰伯特(Leslie Lamport)1990年提出的一种基于消息传递的一致性算法，它曾就此发表了《The Part-Time Parliament》，《Paxos Made Simple》，由于采用故事的方式来解释此算法，感觉还是很难理解。

1.3.2 算法理解

Paxos 算法是分布式一致性算法用来解决一个分布式系统如何就某个值(决议)达成一致的问题。

一个典型的场景是，在一个分布式数据库系统中，如果各节点的初始状态一致，每个节点都执行相同的操作序列，那么他们最后能得到一个一致的状态。为保证每个节点执行相同的命令序列，需要在每一条指令上执行一个”一致性算法”以保证每个节点看到的指令一致。分布式系统中一般是通过多副本来保证可靠性，而多个副本之间会存在数据不一致的情况。所以必须有一个一致性算法来保证数据的一致。

1.3.3 相关概念

在Paxos算法中，有三种角色：

Proposer
Acceptor
Learners

在具体的实现中，一个进程可能同时充当多种角色。比如一个进程可能既是Proposer又是Acceptor又是Learner。Proposer负责提出提案，Acceptor负责对提案作出裁决（accept与否），learner负责学习提案结果。还有一个很重要的概念叫提案（Proposal）。最终要达成一致的value就在提案里。只要Proposer发的提案被Acceptor接受（半数以上的Acceptor同意才行），Proposer就认为该提案里的value被选定了。

Acceptor告诉Learner哪个value被选定，Learner就认为那个value被选定。只要Acceptor接受了某个提案，Acceptor就认为该提案里的value被选定了。为了避免单点故障，会有一个Acceptor集合，Proposer向Acceptor集合发送提案，Acceptor集合中的每个成员都有可能同意该提案且每个Acceptor只能批准一个提案，只有当一半以上的成员同意了一个提案，就认为该提案被选定了。

1.3.4 算法流程

Propser有两个重要属性，提案编号N, 提案V, 简记 Proposer(N, V)。

Acceptor有三个重要属性，响应提案编号ResN, 接受的提案编号AcceptN, 接收的提案AcceptV，间记Acceptor(ResN, AcceptN, AcceptV)。

1、第一阶段: Prepare准备阶段

Proposer： Proposer生成全局唯一且递增的提案编号N,，向所有Acceptor发送Prepare请求，这里无需携带提案内容，只携带提案编号即可, 即发送 Proposer(N, null)。

Acceptor： Acceptor收到Prepare请求后，有两种情况：

如果Acceptor首次接收Prepare请求, 设置ResN=N，同时响应ok
如果Acceptor不是首次接收Prepare请求，则：

若请求过来的提案编号N小于等于上次持久化的提案编号ResN，则不响应或者响应error。

若请求过来的提案编号N大于上次持久化的提案编号ResN, 则更新ResN=N，同时给出响应。响应的结果有两种，如果这个Acceptor此前没有接受过提案，只返回ok。否则如果这个Acceptor此前接收过提案，则返回ok和上次接受的提案编号AcceptN, 接收的提案AcceptV。

2、第二阶段: Accept接受阶段

Proposer： Proposer收到响应后，有两种情况：

如果收到了超过半数响应ok, 检查响应中是否有提案，如果有的话，取提案V=响应中最大AcceptN对应的AcceptV，如果没有的话，V则有当前Proposer自己设定。最后发出accept请求，这个请求中携带提案V。
如果没有收到超过半数响应ok, 则重新生成提案编号N, 重新回到第一阶段，发起Prepare请求。

Acceptor： Acceptor收到accept请求后，分为两种情况：

如果发送的提案请求N大于此前保存的RespN,接受提案，设置AcceptN = N, AcceptV=V, 并且回复ok。
如果发送的提案请求N小于等于此前保存的RespN，不接受，不回复或者回复error。

Proposer： Proposer收到ok超过半数，则V被选定，否则重新发起Prepare请求。

3、第三阶段: Learn学习阶段

Learner: Proposer收到多数Acceptor的Accept后，决议形成，将形成的决议发送给所有Learner。

第2章 Leader选举

接下来我们就来分析一下Leader的选举和数据同步的大体流程和原理，由于图片上传头条教模糊，具体图参考CSDN上

2.1 startLeaderElection

从上面的图可以看出来，当每台服务器启动后，会启动一个QuorumPeer线程，来看看QuorumPeer的start方法

public synchronized void start() {
    if (!getView().containsKey(myid)) {
        throw new RuntimeException("My id " + myid + " not in the peer list");
    }
    //加载数据：数据引擎( ZKDatabase )负责存储/加载/查找数据(基于目录树结构的KV+操作日志+客户端Session);
    loadDataBase();
    //启动网络通信组件：负责维护与客户端的连接(接收客户端的请求并发送相应的响应)
    startServerCnxnFactory();
    try {
        //启动控制台
        adminServer.start();
    } catch (AdminServerException e) {
        LOG.warn("Problem starting AdminServer", e);
        System.out.println(e);
    }
    //开启选举协调者，并执行选举（这个过程是会持续，并不是一次操作就结束了）
    startLeaderElection();
    //执行QuorumPeer线程的run方法
    super.start();
}

本章内容只讲Leader选举，所以我们直接进入startLeaderElection方法

synchronized public void startLeaderElection() {
	...
    //创建Leader选举算法的对象，这个在后面的选举时会用到该对象，为后面的操作做铺垫
    this.electionAlg = createElectionAlgorithm(electionType);
}

protected Election createElectionAlgorithm(int electionAlgorithm){

    switch (electionAlgorithm) {

        case 3:
            //创建QuorumCnxManager对象
            QuorumCnxManager qcm = createCnxnManager();
            QuorumCnxManager oldQcm = qcmRef.getAndSet(qcm);
            if (oldQcm != null) {
                LOG.warn("Clobbering already-set QuorumCnxManager (restarting leader election?)");
                oldQcm.halt();
            }
            QuorumCnxManager.Listener listener = qcm.listener;
            if(listener != null){
                listener.start();
                FastLeaderElection fle = new FastLeaderElection(this, qcm);
                fle.start();
                le = fle;
            } else {
                LOG.error("Null listener when initializing cnx manager");
            }
            break;
        default:
            assert false;
    }
    return le;
}

QuorumCnxManager 内部维护了一系列的队列，用来保存接收到的、待发送的消息以及消息的发送器，除接收队列以外，其他队列都按照SID分组形成队列集合，如一个集群中除了自身还有3台机器，那么就会为这3台机器分别创建一个发送队列，互不干扰。

public QuorumCnxManager(QuorumPeer self,
                        final long mySid,
                        Map<Long,QuorumPeer.QuorumServer> view,
                        QuorumAuthServer authServer,
                        QuorumAuthLearner authLearner,
                        int socketTimeout,
                        boolean listenOnAllIPs,
                        int quorumCnxnThreadsSize,
                        boolean quorumSaslAuthEnabled) {
    //消息接收队列，用于存放其他服务器发送过来的消息
    this.recvQueue = new ArrayBlockingQueue<Message>(RECV_CAPACITY);
    //消息发送队列，用于保存要发送给其他服务器的消息，按照SID进行分组（就是myid）
    this.queueSendMap = new ConcurrentHashMap<Long, ArrayBlockingQueue<ByteBuffer>>();
    //发送器集合，就是存SenderWorker线程的，每个SenderWorker对应一台远程Zookeeper服务器，也是按照SID进行分组的
    this.senderWorkerMap = new ConcurrentHashMap<Long, SendWorker>();
    //最近发送的消息，每个SID保留最近一条消息
    this.lastMessageSent = new ConcurrentHashMap<Long, ByteBuffer>();

    //创建监听器，在外层代码会用到
    listener = new Listener();
    listener.setName("QuorumPeerListener");
}

2.2 listener.start

QuorumCnxManager.Listener ：为了能够相互投票，Zookeeper集群中的所有机器都需要建立起网络连接。QuorumCnxManager在启动时会创建一个ServerSocket来监听Leader选举的通信端口。开启监听后，Zookeeper能够不断地接收到来自其他服务器地创建连接请求，在接收到其他服务器地TCP连接请求时，会进行处理。为了避免两台机器之间重复地创建TCP连接，Zookeeper只允许SID大的服务器主动和其他机器建立连接，否则断开连接。在接收到创建连接请求后，服务器通过对比自己和远程服务器的SID值来判断是否接收连接请求，如果当前服务器发现自己的SID更大，那么会断开当前连接，然后自己主动和远程服务器将连接（自己作为“客户端”）。一旦连接建立，就会根据远程服务器的SID来创建相应的消息发送器SendWorker和消息发送器RecvWorker，并启动。

public void run() {
   
    while ((!shutdown) && (portBindMaxRetry == 0 || numRetries < portBindMaxRetry)) {
        try {
            while (!shutdown) {
                //参与投票
                receiveConnection(client);
            }
        }
    }
}

receiveConnection——>handleConnection
在上面的方法中会拿到别的服务器发过来的myid

private void handleConnection(Socket sock, DataInputStream din)
    throws IOException {
    Long sid = null, protocolVersion = null;
    InetSocketAddress electionAddr = null;

    try {
        protocolVersion = din.readLong();
        if (protocolVersion >= 0) { // this is a server id and not a protocol version
            sid = protocolVersion;
        } else {
            try {
                //获取sid，就是myid的值
                InitialMessage init = InitialMessage.parse(protocolVersion, din);
                sid = init.sid;
                electionAddr = init.electionAddr;
            } catch (InitialMessage.InitialMessageException ex) {
               
            }
        } 
    } 
}

然后回到QuorumPeer的方法createElectionAlgorithm中，在启动一些列的线程之后，会创建一个leader选举算法的实例，FastLeaderElection fle = new FastLeaderElection(this, qcm);

2.3 FastLeaderElection

public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager){
    this.stop = false;
    this.manager = manager;
    //跟进
    starter(self, manager);
}

创建FastLeaderElection会调用starter()方法，该方法会创建sendqueue、recvqueue队列、Messenger对象，其中Messenger对象的作用非常关键，方法源码如下：

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

创建Messenger的时候，会创建WorkerSender并封装成wsThread线程，创建WorkerReceiver并封装成wrThread线程，看名字就很容易理解，wsThread用于发送数据，wrThread用于接收数据，创建完FastLeaderElection后接着会调用它的start()方法启动选举算法，实际上就是启动这两个用于数据通信的线程。下面我们来分别看看这两个线程的逻辑：

wsThread由WorkerSender封装而来，run方法会调用process()方法，源码如下：

void process(ToSend m) {
    ByteBuffer requestBuffer = buildMsg(m.state.ordinal(), m.leader, m.zxid, m.electionEpoch, m.peerEpoch, m.configData);
    //把对应的sid作为了消息发送出去，这里其实是发送投票信息
    manager.toSend(m.sid, requestBuffer);
}

投票可以投自己，也可以投别人，如果是选票选自己，只需要把投票信息添加到recvQueue中即可，源码如下：

public void toSend(Long sid, ByteBuffer b) {

    //如果投票给自己，只需要吧信息添加到自身的RecvQueue中
    if (this.mySid == sid) {
        b.position(0);
        addToRecvQueue(new Message(b.duplicate(), sid));
    } else {
        //投票给别人则把信息添加到SendQueue，并建立连接
        ArrayBlockingQueue<ByteBuffer> bq = new ArrayBlockingQueue<ByteBuffer>(
            SEND_CAPACITY);
        ArrayBlockingQueue<ByteBuffer> oldq = queueSendMap.putIfAbsent(sid, bq);
        if (oldq != null) {
            addToSendQueue(oldq, b);
        } else {
            addToSendQueue(bq, b);
        }
        //这个方法里面也会创建两个线程，把投给别人的票的信息添加到SendQueue中，同时也添加到RecvQueue中
        connectOne(sid);
    }
}

在来看看处理接收数据的线程，在WorkerReceiver.run方法中会从recvQueue中获取Message，并把发送给其他服务的投票封装到sendqueue队列中，交给WorkerSender发送处理，源码太长就不贴出来了。

到这一步，实际上前面做的工作都是为了进行选举做的一些列的准备，包括创建监听器，创建FastLeaderElection对象，都是在创建一些线程保证服务节点之间的通信，当然，还提供了选举的核心算法，但是目前代码还没有调到选举的算法中去。

在哪里调的呢？我们接着我下走。程序回到QuorumPeer的start方法中。以上的流程都是在startLeaderElection方法中完成的，接下来，最后一步是调用super.start方法，实际上就是启动本上QuorumPeer这个线程，最后执行到QuorumPeer.run。

2.4 Leader选举核心

所有节点初始状态都为LOOKING，会进入到选举流程，选举流程首先要获取算法，获取算法的方法是makeLEStrategy()，该方法返回的是FastLeaderElection实例，核心选举流程是FastLeaderElection中的lookForLeader()方法。

try {
    reconfigFlagClear();
    if (shuttingDownLE) {
        shuttingDownLE = false;
        startLeaderElection();
    }
    //makeLEStrategy是获取之前创建的FastLeaderElection的对象
    // 选举的核心方法lookForLeader
    setCurrentVote(makeLEStrategy().lookForLeader());
} catch (Exception e) {
    LOG.warn("Unexpected exception", e);
    setPeerState(ServerState.LOOKING);
}

lookForLeader()是选举过程的关键流程，源码分析如下：

public Vote lookForLeader() throws InterruptedException {
    ...
    try {
        //接收的投票
        HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();

        //对外的投票记录
        HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();

        synchronized(this){
            //投票轮次自增
            logicalclock.incrementAndGet();
            //首次推举自己为leader
            updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
        }
        //广播发出投票
        sendNotifications();

        //如果当前server状态为LOOKING，且未选出leader则一直循环
        while ((self.getPeerState() == ServerState.LOOKING) &&
               (!stop)){
            //准备接收发来的投票，该过程数据阻塞过程，直到超时
            Notification n = recvqueue.poll(notTimeout,
                                            TimeUnit.MILLISECONDS);

            //如果没有接收到投票，则继续发出广播进行投票，否则处理投票信息
            if(n == null){
                //如果集群中有其他节点信息，就开始广播
                if(manager.haveDelivered()){
                    //开始广播自己的投票信息
                    sendNotifications();
                } else {
                    //如果服务器不存在，尝试与每个服务器建立连接
                    manager.connectAll();
                }

                //延长从队列获取选票的时长
                int tmpTimeOut = notTimeout*2;
                notTimeout = (tmpTimeOut < maxNotificationInterval?
                              tmpTimeOut : maxNotificationInterval);
            } 
            else if (validVoter(n.sid) && validVoter(n.leader)) {
     
                switch (n.state) {
                        //如果当前选举人是LOOKING状态
                    case LOOKING:
                        //如果收到服务器轮次大于自己的，则将自己的轮次设置成最新的，将自己的投票池清空
                        if (n.electionEpoch > logicalclock.get()) {
                            logicalclock.set(n.electionEpoch);
                            recvset.clear();
                            //进行选票PK，如果自己的票没有PK过其他投递的票，则将自己的票变更为其他
                            if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                                   getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
                                updateProposal(n.leader, n.zxid, n.peerEpoch);
                            } else {
                                ...
                            }
                            //重新发出投票
                            sendNotifications();
                        } else if (n.electionEpoch < logicalclock.get()) {
                            //如果收到的轮次小于自己的轮次，不搭理他，他的是无效的
                            if(LOG.isDebugEnabled()){
                                ...
                            }
                            break;
                        } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                                       proposedLeader, proposedZxid, proposedEpoch)) {
                            //如果收到的轮次刚好等于自己的，则进行选票PK，如果自己的票没有PK过其他投递的票，则将自己的票变更为其他
                            updateProposal(n.leader, n.zxid, n.peerEpoch);
                            //重新发出投票
                            sendNotifications();
                        }

                        //将获得的投票数据放入到自己的票池中
                        recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));

                        //判断是否超过半数的票指向同一个服务ID（核心）
                        if (termPredicate(recvset,
                                          new Vote(proposedLeader, proposedZxid,
                                                   logicalclock.get(), proposedEpoch))) {

                            //如果此刻在票池汇总还有未取出的投票，则和选举出的投票PK，如果取出的票优于当前推举的投票，则重新投
                            while((n = recvqueue.poll(finalizeWait,
                                                      TimeUnit.MILLISECONDS)) != null){
                                if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                                       proposedLeader, proposedZxid, proposedEpoch)){
                                    recvqueue.put(n);
                                    break;
                                }
                            }

                            //如果票池中没有可以PK的投票，则就认为选举出来的服务为leader
                            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:
                        LOG.debug("Notification from observer: " + n.sid);
                        break;
                    case FOLLOWING:
                    case LEADING:
                        //如果新收到的选票发送者角色是Leader且选票轮次和自己的选票轮次一样
                        if(n.electionEpoch == logicalclock.get()){
                            //将leader角色投递的这张选票放入自己的选票池中
                            recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
                            //判断是否有超过半数的票数是推荐了n推荐的leader，且n.leader也确实是LEADING状态
                            if(termPredicate(recvset, new Vote(n.version, n.leader,
                                                               n.zxid, n.electionEpoch, n.peerEpoch, n.state))
                               && checkLeader(outofelection, n.leader, n.electionEpoch)) {
                                //指定n推荐的为真正的leader，同时修改其他服务对应的状态
                                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;
                            }
                        }

                        //如果轮次不一致，则将n的投票记录到outofelection中
                        outofelection.put(n.sid, new Vote(n.version, n.leader, 
                                                          n.zxid, n.electionEpoch, n.peerEpoch, n.state));
                        //判断是否有超过半数的票数是推荐了n推荐的leader，且n.leader也确实是LEADING状态
                        if (termPredicate(outofelection, new Vote(n.version, n.leader,
                                                                  n.zxid, n.electionEpoch, n.peerEpoch, n.state))
                            && checkLeader(outofelection, n.leader, n.electionEpoch)) {
                            synchronized(this){
                                //更新当前服务选举轮次
                                logicalclock.set(n.electionEpoch);
                                //指定n推荐的为真正的leader，同时修改其他服务对应的状态
                                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:
                        ...
                }
            } else {
                ...
            }
        }
        return null;
    } finally {
        ...
    }
}

上面这个流程其实也不难，就是每次拿到票了之后，先判断这张票的轮次跟我的是不是一样的或者大于我自身的轮次，如果是或者大于，则就看这张票投的那个人和我要投的那个人哪个牛逼一点，谁牛逼最终就投给谁。完事之后，把别人投的这张票放到一个池子里面，然后通过过半函数去进行判断，判断选举的这个人是否满足过半条件，且它的状态是LEADING，如果满足，完事，他是Leader了。

那么接下来我们再来看看过半半数是怎么判断的。termPredicate代码如下：

protected boolean termPredicate(Map<Long, Vote> votes, Vote vote) {

    //过半的判断
    return voteSet.hasAllQuorums();
}

public boolean hasAllQuorums() {
    for (QuorumVerifierAcksetPair qvAckset : qvAcksetPairs) {
        //算法过程
        if (!qvAckset.getQuorumVerifier().containsQuorum(qvAckset.getAckset()))
            return false;
    }
    return true;
}

QuorumMaj.containsQuorum

public QuorumMaj(Map<Long, QuorumServer> allMembers) {
    this.allMembers = allMembers;
    for (QuorumServer qs : allMembers.values()) {
        if (qs.type == LearnerType.PARTICIPANT) {
            votingMembers.put(Long.valueOf(qs.id), qs);
        } else {
            observingMembers.put(Long.valueOf(qs.id), qs);
        }
    }
    half = votingMembers.size() / 2;
}

public boolean containsQuorum(Set<Long> ackSet) {
    return (ackSet.size() > half);
}

投票规则

我们来看一下选票PK的方法totalOrderPredicate()，该方法其实就是Leader选举规则，规则有如下三个：

比较 epoche(zxid高32bit)，如果其他节点的epoche比自己的大，选举 epoch大的节点（理由：epoch 表示年代，epoch越大表示数据越新）代码：(newEpoch > curEpoch)；
比较 zxid，如果epoche相同，就比较两个节点的zxid的大小，选举 zxid大的节点（理由：zxid 表示节点所提交事务最大的id，zxid越大代表该节点的数据越完整）代码：(newEpoch == curEpoch) && (newZxid > curZxid)；
比较 serviceId，如果 epoch和zxid都相等，就比较服务的serverId，选举 serviceId大的节点（理由： serviceId 表示机器性能，他是在配置zookeeper集群时确定的，所以我们配置zookeeper集群的时候可以把服务性能更高的集群的serverId设置大些，让性能好的机器担任leader角色）代码：(newEpoch == curEpoch) && ((newZxid == curZxid) && (newId > curId))。

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

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

至此，Leader选举分析完了，接再来我们在看看，Zookeeper集群中的数据是如何同步的。

第3章数据同步

3.1 同步流程

当Zookeeper服务端启动后，会先进行角色判断，如果是looking的话，则发起leader选举，如果是leader的话，则调用leader.lead方法，创建一个接收连接的LearnerCnxAcceptor线程，在LearnerCnxAcceptor线程内部又建立一个阻塞的LearnerHandler线程等待别的节点的连接。

如果是follower的话，则调用follower.followLeader方法首先查找leader的Socket服务端，然后建立连接。当follower建立连接后，leader端会建立一个LearnerHandler线程相对应，用来处理follower与leader的数据包传输。下面就是follow与leader进行数据包传输的具体流程：

follower端封装当前zk服务器的Zxid和Leader.FOLLOWERINFO的LearnerInfo数据包发送给leader
leader端这时处于getEpochToPropose方法的阻塞时期，需要得到Learner端超过一半的服务器发送Epoch
getEpochToPropose解阻塞之后，LearnerHandler线程会把超过一半的Epoch与leader比较得到最新的newLeaderZxid，并封装成Leader.LEADERINFO包发送给Learner端
Learner端得到最新的Epoch，会更新当前服务器的Epoch。并把当前服务器所处的lastLoggedZxid位置封装成Leader.ACKEPOCH发送给leader
此时leader端处于waitForEpochAck方法的阻塞时期，需要得到Learner端超过一半的服务器发送EpochACK
当waitForEpochAck阻塞之后便可以在LearnerHandler线程内决定用那种方式进行同步。如果Learner端的lastLoggedZxid>leader端的，Learner端将会被删除多余的部分。如果小于leader端的，将会以不同方式进行同步
leader端发送Leader.NEWLEADER数据包给Learner端（6、7步骤都是另开一个线程来发送这些数据包）
Learner端同步之后，会在一个while循环内处理各种leader端发送数据包，包括两阶段提交的Leader.PROPOSAL、Leader.COMMIT、Leader.INFORM等。在同步数据后会处理Leader.NEWLEADER数据包，然后发送Leader.ACK给leader端
此时leader端处于waitForNewLeaderAck阻塞等待超过一半节点发送ACK。

大体流程就是这样，接下来我们来看看具体源码

3.2 Follow同步

随着QuorumPeerMain启动，会创建QuorumPeer线程，并执行该线程的run方法，在run方法中判断节点的角色，如果是Follower的话，执行下面的代码

case FOLLOWING:
    try {
        LOG.info("FOLLOWING");
        //创建Follower对象，并设置Follower
        setFollower(makeFollower(logFactory));
        /**
         * 该方法中主要的逻辑有2个
         * 1.跟Leader建立连接
         * 2.通过一个死循环与Leader进行数据同步，同步完毕后，正常接收Leader的请求，并且执行对应的逻辑，包括Request、Propose、Commit请求
         */
        follower.followLeader();
    } catch (Exception e) {
        LOG.warn("Unexpected exception",e);
    } finally {
        //如果集群超过半数服务宕机或者Leader宕机，首先设置调用shutdown，然后设置状态为looking，触发重新选举，因为当前这个case操作的外层也是一个死循环
        follower.shutdown();
        setFollower(null);
        updateServerState();
    }
break;

核心代码在follower.followLeader()中，接下来我们再往底层走

3.2.1 followLeader

void followLeader() throws InterruptedException {
        ...
        try {
            //首先寻找Leader的服务器
            QuorumServer leaderServer = findLeader();            
            try {
                //跟Leader建立连接
                connectToLeader(leaderServer.addr, leaderServer.hostname);
                //注册Follower，会将当前Follower节点信息发送给Leader节点
                long newEpochZxid = registerWithLeader(Leader.FOLLOWERINFO);
               	...
                //和Leader节点同步历史数据
                syncWithLeader(newEpochZxid);                
                QuorumPacket qp = new QuorumPacket();
                //通过死循环与Leader进行数据同步
                while (this.isRunning()) {
                    //读取数据包
                    readPacket(qp);
                    //处理数据包
                    processPacket(qp);
                }
            } catch (Exception e) {
               ...
            }
        } finally {
            ...
        }
    }

从这段代码可以看出，followLeader方法就干了以下几件事

寻找Leader并和Leader建立连接
向Leader注册Follower，会将当前Follower节点信息发送给Leader节点
和Leader同步历史数据
读取Leader发送的数据包并同步Leader数据包

针对这四件事，我们分别来看看怎么完成的

3.2.2 findLeader

protected QuorumServer findLeader() {
    QuorumServer leaderServer = null;
    // Find the leader by id
    //找到当前投票选出来的Leader
    Vote current = self.getCurrentVote();
    //循环匹配Leader
    for (QuorumServer s : self.getView().values()) {
        if (s.id == current.getId()) {
            // Ensure we have the leader's correct IP address before
            // attempting to connect.
            s.recreateSocketAddresses();
            leaderServer = s;
            break;
        }
    }
    if (leaderServer == null) {
        LOG.warn("Couldn't find the leader with id = "
                 + current.getId());
    }
    return leaderServer;
}

建立连接的方法就不看了，实际上就是通过socket去建立的

3.2.3 registerWithLeader

当拿到了Leader节点的信息并建立连接之后，Follower会将自身的信息注册到Leader节点上去，这一步操作就是为了保证后续Leader要同步数据的时候知道要同步给谁。

protected long registerWithLeader(int pktType) throws IOException{

    ...
    //向Leader写payLoad
    boa.writeRecord(li, "LearnerInfo");
    qp.setData(bsid.toByteArray());
	//向Leader写数据包
    writePacket(qp, true);
    readPacket(qp);        
    final long newEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());
    if (qp.getType() == Leader.LEADERINFO) {
        ...
        //向Leader写ackNewEpoch
        QuorumPacket ackNewEpoch = new QuorumPacket(Leader.ACKEPOCH, lastLoggedZxid, epochBytes, null);
        writePacket(ackNewEpoch, true);
        return ZxidUtils.makeZxid(newEpoch, 0);
    } else {
       ...
    }
}

3.2.4 readPacket

这个方法就是读取Leader的数据包的封装，源码如下：

void readPacket(QuorumPacket pp) throws IOException {
    synchronized (leaderIs) {
        //读取Leader的数据
        leaderIs.readRecord(pp, "packet");
    }
 	...
}

3.3 Leader同步

3.3.1 lead

如果当前节点的状态是LEADING的话，则会执行leader.lead()，leader.lead()是执行的核心业务流程，源码如下：

case LEADING:
    LOG.info("LEADING");
    try {
        //创建Leader对象，并设置Leader
        setLeader(makeLeader(logFactory));
        /**
        * 同样有个死循环，主要的工作是：
        * 1、跟Follower建立连接
        * 2、跟Follower进行数据同步
        * 3、还要能正常接收客户端zkCli的请求
        */
        leader.lead();
        //如果集群半数挂了，则先设置Leader对象为null
        setLeader(null);
    } catch (Exception e) {
        LOG.warn("Unexpected exception",e);
    } finally {
        if (leader != null) {
            leader.shutdown("Forcing shutdown");
            setLeader(null);
        }
        //这是状态为LOOKING，触发Leader选举
        updateServerState();
    }
break;

leader.lead()方法是Leader执行的核心业务流程，源码过长，简化源码如下：

void lead() throws IOException, InterruptedException {
    ...
    try {
        //从快照和事务日志中加载数据
        zk.loadData();

        //创建一个线程，接收Follower或者Observer的连接
        cnxAcceptor = new LearnerCnxAcceptor();
        //开启线程
        cnxAcceptor.start();

        //等待超过一半的Follower和Observer连接，这里才会往下执行，返回新的轮次epoch
        //否则阻塞
        long epoch = getEpochToPropose(self.getId(), self.getAcceptedEpoch());
        //根据新的epoch，设置新的起始zxid
        zk.setZxid(ZxidUtils.makeZxid(epoch, 0));

        //等待超过一半的Follower和Observer获取新的epoch，并且返回Leader.ACKEPOCH
        //这里才会往下执行，否则阻塞
        waitForEpochAck(self.getId(), leaderStateSummary);
        //设置最新的轮次
        self.setCurrentEpoch(epoch);    

        try {
            //等待超过一半的Follower和Observer进行数据同步，并且返回Leader.ACK
            //这里才会往下执行，否则阻塞
            waitForNewLeaderAck(self.getId(), zk.getZxid());
        } catch (InterruptedException e) {
           ...
            return;
        }
        //走到这里说明集群中数据已经同步成功，可以正常运行了
        //开启zkServer，并且同时开启请求调用链接收业务请求
        startZkServer();

        //进行死循环。每次休眠self.tickTime/2，同时对所有的Follower和Observer发起心跳检测
        while (true) {
            synchronized (this) {
                long start = Time.currentElapsedTime();
                long cur = start;
                long end = start + self.tickTime / 2;
                while (cur < end) {
                    wait(end - cur);
                    cur = Time.currentElapsedTime();
                }

                if (!tickSkip) {
                    self.tick.incrementAndGet();
                }

                //将Follower加入该容器
                syncedAckSet.addAck(self.getId());

                for (LearnerHandler f : getLearners()) {
                    if (f.synced()) {
                        syncedAckSet.addAck(f.getSid());
                    }
                }

                // check leader running status
                if (!this.isRunning()) {
                    // set shutdown flag
                    shutdownMessage = "Unexpected internal error";
                    break;
                }

                //判断是否有超过一半的Follower在集群中
                if (!tickSkip && !syncedAckSet.hasAllQuorums()) {
                    // Lost quorum of last committed and/or last proposed
                    // config, set shutdown flag
                    //如果没有。则调用shutdown关闭对象，并重新选举
                    shutdownMessage = "Not sufficient followers synced, only synced with sids: [ "
                        + syncedAckSet.ackSetsToString() + " ]";
                    break;
                }
                tickSkip = !tickSkip;
            }
            for (LearnerHandler f : getLearners()) {
                f.ping();
            }
        }
        //如果没有。则调用shutdown关闭对象，并重新选举
        if (shutdownMessage != null) {
            shutdown(shutdownMessage);
            // leader goes in looking state
        }
    } finally {
        zk.unregisterJMX(this);
    }
}

该方法的核心功能做个总结：

从快照和事务日志中加载数据
创建一个线程，接收Follower/Observer的连接
等待超过一半的(Follower和Observer)连接，再继续往下执行程序
等待超过一半的(Follower和Observer)获取了新的epoch，并且返回了Leader.ACKEPOCH，再继续往下执行程序
等待超过一半的(Follower和Observer)进行数据同步成功，并且返回了Leader.ACK，再继续往下执行程序
数据同步完成，开启zkServer，并且同时开启请求调用链接收请求执行
进行一个死循环，每次休眠self.tickTime / 2，和对所有的(Observer/Follower)发起心跳检测
集群中没有过半Follower在集群中，调用shutdown关闭一些对象，重新选举

3.3.2 LearnerCnxAcceptor

在该方法中会创建一个接收followers连接的LearnerCnxAcceptor线程，根据配置的同步的地址的数量（例如：server.2=127.0.0.1:12881:13881 配置同步的端口是12881只有一个）阻塞等待连接，run方法源码如下：

public void run() {

    while (!stop) {
        Socket s = null;
        boolean error = false;

        //根据配置获取follower的连接地址，并阻塞等待Follower来连接
        s = ss.accept();
		//创建LearnerHandler线程
        LearnerHandler fh = new LearnerHandler(s, is, Leader.this);
        //执行该线程
        fh.start();
    }
}

LearnerCnxAcceptor的run方法中创建了LearnerHandler对象，在接收到连接后，就会调用LearnerHandler，它的run方法就是读取或写数据包与Learner交换数据包。如果没有数据包读取，则会阻塞当前方法ia.readRecord(qp, "packet");，源码如下：

下文预告

Zookeeper分布式锁原理分析
Zookeeper用作Leader选举的原理分析

网站首页 > 技术文章正文

Zookeeper系列——5Zookeeper的数据一致性分析

学习目标

第1章集群简介

1.1 集群角色

1.2 ZAB协议

1.2.1 概念

1.2.2 结论

1.3 Paxos算法

1.3.1 介绍

1.3.2 算法理解

1.3.3 相关概念

1.3.4 算法流程

第2章 Leader选举

2.1 startLeaderElection

2.2 listener.start

2.3 FastLeaderElection

2.4 Leader选举核心

第3章数据同步

3.1 同步流程

3.2 Follow同步

3.2.1 followLeader

3.2.2 findLeader

3.2.3 registerWithLeader

3.2.4 readPacket

3.3 Leader同步

3.3.1 lead

3.3.2 LearnerCnxAcceptor

下文预告

猜你喜欢

网站首页 > 技术文章 正文

Zookeeper系列——5Zookeeper的数据一致性分析

学习目标

第1章 集群简介

1.1 集群角色

1.2 ZAB协议

1.2.1 概念

1.2.2 结论

1.3 Paxos算法

1.3.1 介绍

1.3.2 算法理解

1.3.3 相关概念

1.3.4 算法流程

第2章 Leader选举

2.1 startLeaderElection

2.2 listener.start

2.3 FastLeaderElection

2.4 Leader选举核心

第3章 数据同步

3.1 同步流程

3.2 Follow同步

3.2.1 followLeader

3.2.2 findLeader

3.2.3 registerWithLeader

3.2.4 readPacket

3.3 Leader同步

3.3.1 lead

3.3.2 LearnerCnxAcceptor

下文预告

猜你喜欢

网站首页 > 技术文章正文

第1章集群简介

第3章数据同步