zookeeper的灵魂-zab协议

​ 之前每次面试前，都要把zookeeper的相关知识再看一遍，这次直接把对zookeeper的一些理解记录下来，方便查阅。
​ 由于zookeeper涉及到的东西比较多，所以分成两部分介绍，本次写的是zookeeper的核心-zab数据一致性协议。

1 zk-选举流程

zookeeper的zab协议主要包括两部分，一是leader的选举，二是数据的同步过程。

在zookeeper中节点（在代码中称为Quorum）类型有三种

  1. leader 选举过程中得到集群中半数以上节点认同的节点。
  2. follwer follwer参与选举过程，其中follwer与observer在代码中都表示为Learner。
  3. observer observer是一种节点的启动类型，表示该节点不会参与选举。
  4. read-only 准确的讲并不是在选举中需要用到的，而是一种启动类型，如果以read-only模式启动，当zookeeper集群发生故障不可用，该节点依然能够提供读服务。

zookeeper中的数据都保存在每个节点的内存中（也就是冗余储存，每个机器一份），每个操作会有TxnLog记录，并且会有定时的合并成snapshot文件，存储在磁盘上，保证数据不丢失。

当zookeeper集群启动时，每台机器要先从磁盘中恢复数据到内存中。并获取到各自数据的最后一条事务操作id-lastProcessedZxid，lastProcessedZxid其实意味着每台机器的数据完整程度，lastProcessedZxid越大表示该台机器的数据最完整，zk的选举主要就是选择一台lastProcessedZxid最大的机器当做leader，这样在后续的节点数据同步时，工作量就会少很多。加载完数据以后，根据config中的集群列表，与每个节点建立通讯(默认是NIO，可以加参数配置选择netty)，并将io连接保存下来，由QuorumCnxManager类统一管理，之后就进入选举流程。

选举代码中一些重要的数据结构的描述

节点选举过程中发送的消息结构

1
2
3
4
5
6
7
8
9
10
static public class ToSend {
    static enum mType {crequest, challenge, notification, ack} // 选举发送的为notification
    long leader; // 推举的leaderId (sid)
    long zxid;   // 初始化为从磁盘恢复的lastProcessedZxid,前32位表示peerEpoch，后32位为leader的提议计数
    long electionEpoch;  // 选举轮次，每次投票都会+1  与logicalclock概念一致
    QuorumPeer.ServerState state;  // 当前节点状态，初始化为LOOKING
    long sid;   // 发送方的sid
    long peerEpoch;      // leader当选的轮次,每一次选举成功后+1
  	byte[] configData;  // 集群的快照 对应QuorumVerifier类
}

选票的PK算法totalOrderPredicate逻辑：

​ 先比较两个选票的peerEpoch，大者胜利

​ 再比较zxid的大小，大者胜利

​ 再比较sid的大小，大者胜利

确认选举完成的算法termPredicate逻辑：

​ 每张选票所选举的sid，是否得到了集群中超过半数的票数。

zk选举过程流程图大致如下

zab

  1. 刚开始选举时，先投自己一票

  2. 接收Quorum-A的投票n，并判断A在不在集群中。

  3. 判断选票轮次（用自己的logicalclock和选票n中的electionEpoch比较）

    如果自己的logicalclock大，忽略掉该选票，继续等待接收投票

    如果投票中的electionEpoch大，表名自己已经落后，更新自己的logicalclock，并清除票箱中的投票，把自己的票投给A发送出去。

  4. 如果选票轮次一致，进行选票PK，也就是totalOrderPredicate方法，比较zxid和myid的大小，谁获取，把票投给谁，并把最新的票发送给其他节点。

  5. 把更改后的选票加入到票箱中，判断选举过程是否能够结束（是否有多数节点都投给一个节点）

选举结束以后的数据同步

前面提到过，选举最核心的是选一个zxid最大的当leadre，这样在结束以后，集群各个节点同步数据时，大多数都是发送DIFF，补充缺失的数据就好了，减少同步数据的处理时间和复杂性。

当选举完成后，leader向learner节点发送确认通知，learner节点在返回ack前，会调用syncFollower方法与leader对比数据。

其中数据对比的可能出现情况如下：

  1. leader.lastProcessedZxid == peerLastZxid leader的数据和follwer的数据一致，发送DIFF类型数据包

  2. leader.maxCommittedLog < peerLastZxid follower的zxid > leader的zxid，follower节点要把多余的数据删除掉,发送TRUNC类型数据包

  3. leader.maxCommittedLog >= peerLastZxid && leader.minCommittedLog <= peerLastZxid

    follower的数据略少于leader的数据，发送txnlog + committedLog包进行同步

  4. leader.minCommittedLog > peerLastZxid follower的数据远少于leader的数据，发送SNAP类型数据包，并直接发送snapshot文件让follower同步

当数据同步完成后，集群此刻才进入可用状态。

2 zab-如何保证数据一致性

zk可以保证请求的一致性为：每个session的请求按顺序执行、写请求按照zxid顺序执行、确认一个session中写请求之间没有竞争

当集群正常工作时，zab协议同时负责每个节点的数据同步工作。

所以现在重点来写一下，当一个写请求提交时，zk集群的响应流程主要如下图所示。

zk-write-process

  1. PrepRequestProcessor

    所有请求类型第一层要经过的处理链。

    主要职责是对request请求进行初始化，如果是事务请求则设置相应的请求体，并将写操作记录到outstandingChanges与outstandingChangesForPath队列中（对这两个队列的操作都会使用outstandingChanges加对象锁，保证并发安全）。非事务请求只检查session是否有效。(Watche的操作也都属于非事务请求)

  2. ProposalRequestProcessor

    ProposalRequestProcessor的下一条链是CommitProcessor，在初始化时，会先启动SyncRequestProcessor。该请求链的主要职责是，收到请求后，调用Leader的propose方法，构造提议消息体，并将该提议发送给集群内的其他节点。

    向CommitProcessor的queuedRequests队列入队请求。

    之后，再向SyncRequestProcessor的queuedRequests队列中入队本次的请求。

  3. SyncRequestProcessor

    SyncRequestProcessor的主要职责是确保ProposalRequestProcessor传递的请求能够保存到本地磁盘上，在SyncRequestProcessor中有线程不断的轮询queuedRequests队列，接收到Request，就将Request写到事务日志文件中。如果写入失败，则请求失败。

  4. AckRequestProcessor

    在Leader节点处理过程中，SyncRequestProcessor会调用AckRequestProcessor，向自己发送ACK确认信息。

  5. CommitProcessor

    CommitProcessor的主要职责是检查事务请求是否能够被提交。(如果是非事务类型请求，就直接放行了，进入下一条请求链处理)

    CommitProcessor是最复杂的一个请求链，同时和ProposalRequestProcessor、SyncRequestProcessor有交互（还都是用队列，异步进行的，所以理解起来非常难= =）。

    先看一下内部重要的数据结构

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    // 所有请求通过processPacket方法首先放入该队列，等待后续处理
    protected final LinkedBlockingQueue<Request> queuedRequests =
        new LinkedBlockingQueue<Request>();
    // 被集群半数节点通过的请求，会到这个队列里面
    protected final LinkedBlockingQueue<Request> committedRequests =
        new LinkedBlockingQueue<Request>();
    // 下一个等待提交的请求
    protected final AtomicReference<Request> nextPending =
            new AtomicReference<Request>();
    // 当前正在执行的请求
    private final AtomicReference<Request> currentlyCommitting =
            new AtomicReference<Request>();

    该请求链的主循环不断判断有没有待提交的请求，如果有，调用WorkerService提交请求执行到下一请求链。WorkerService用于保证同一session的请求能够被顺序发送。

    Follower在接收到leader的提议时，会先保存zxid，等再接收到commit请求后会调用commit方法进行处理提交的写请求，先对比pendingTxns的zxid和commit的zxid，如果不一致则follwer直接退出，重新同步leader数据。

  6. ToBeAppliedRequestProcessor

    CommitProcessor的下一条链，该链的下一条链为FinalRequestProcessor。

    该请求链中有个待应用队列toBeApplied

    1
    private final ConcurrentLinkedQueue<Proposal> toBeApplied = new ConcurrentLinkedQueue<Proposal>();

    当Leader收到集群节点的大多数ACK时，会把请求放入到toBeApplied队列中。

    当本请求链被调用时，先直接调用FinalRequestProcessor，完成事务请求的落地，再从toBeApplied队列中删除相应的事务请求。

  7. FinalRequestProcessor

    所有请求的最终处理链，主要职责是负责将事务请求落实到内存数据库中，也就是写库操作。写完库以后会把outstandingChanges相应的record删除掉，并向follower发送数据的comitted数据包。