Zookeeper源码分析

Zookeeper的启动流程

1. zookeeper的主类是QuorumPeerMain，启动时读取zoo.cfg配置文件，如果没有配置server列表，则单机模式启动，否则按集群模式启动，这里只分析集群模式

2. 根据配置初始化quorumPeer对象，并启动quorumPeer线程，这里主要做了几件事情
   1. 读取保存在磁盘上的数据，包括db的snapshot和txnlog，zookeeper的存储结构另外专门讲述
   2. 启动cnxnFactory，这里主要是启动一个server，用来接收来自client的请求，绑定在配置文件中的clientPort端口
   3. 在QuorumAddress上绑定一个server，用来和其他zookeeper server做交互
   4. 启动leader选举过程，因为server刚启动时是存在LOOKING状态，需要发一起一次选举过程来获取leader
      
   5. 启动quorumPeer的主线程run，根据当前节点的状态来启动不同的流程
      1. 如果是Looking状态，则调用FastLeaderElection::lookForLeader来发起选举流程
      2. 如果是OBSERVING状态，则开始Observer流程
      3. 如果是FOLLOWING状态，则开始Follower流程
      4. 如果是LEADING状态，则开始Leader流程

Zookeeper的选举流程 

1. 发起选举流程有两种情况：

        1）server刚启动的时候，server的状态初始化为LOOKING状态

        2）server发生异常，切换到LOOKING状态

        server有4中状态

1. LOOKING：初始状态，表示在选举leader
2. FOLLOWING：跟随leader的角色，参与投票
3. LEADING：集群的leader
4. OBSERVING：不参与投票，只是同步状态

2. 按刚启动来讲述选举流程，QuorumPeer::start() -> QuorumPeer::startLeaderElection() -> QuorumPeer::createElectionAlgorithm

3. 默认使用FastLeaderElection算法，初始化的流程如下：
   1. 初始化QuorumCnxManager，管理选举中和其他server的交互，选举时监听在专门的electionAddr上
   2. QuorumCnxManager是实际发生网络交互的地方，它的主要数据结构包括
      1. queueSendMap：sid -> buffer queue，为每个参与投票的server都保留一个队列
      2. recvQueue：message queue，所有收到的消息都放到recvQueue
      3. listener：server主线程，收发消息时和上面两个队列交互

QuorumCnxManager可以保证每对peer之间只有一个链接，如果有server发起新的链接，则比较sid，sid大的保留链接，小的放弃链接

3. 初始化FastLeaderElection，这是选举逻辑所在的地方，它主要包括3个线程：
   1. Messenger::WorkerReceiver：从QuorumCnxManager::recvQueue中获取网络包，并将其发到FastLeaderElection::recvqueue中
   2. Messenger::WorkerSender：从FastLeaderElection::sendqueue中获取网络包，并将其放到QuorumCnxManager::queueSendMap中，并发送到网络上
   3. lookForLeader：QuorumPeer主线程会调用lookForLeader函数，它从recvqueue中获取别人发给server的选举数据，并将发给其他server的投票放到sendqueue中

4. FastLeaderElection::lookForLeader中实现了选举算法，具体的流程如下：
   1. 首先更新选举周期logicalclock，并把自己作为leader作为投票发给所有其他的server
   2. 然后进入本轮投票的循环
      1. 从recvqueue获取一个网络包，如果没有收到包则检查是否要重连和重发自己的投票
      2. 收到投票后判断投票的状态
         1. LOOKING：
            1. 如果对方投票的周期大于自己的周期，那就清空自己的已经收到的投票集合recvset，并将自己作为候选和对方投票的leader做比较，选出大的作为新的投票，然后再发送给所有人。

这里比较大小是通过比较（zxid，sid）这个二元组来的，zxid大的就大，否则sid大的就大

2. 如果对方的投票周期小于自己，则忽略对方的投票

3. 如果周期相等，则比较对方的投票和自己认为的候选，选出大的作为新的候选，然后再发送给所有人

4. 然后判断当前收到的投票是否可以得出谁是leader的结论，这里主要是通过判断当前的候选leader在收到的投票中是否占了多数

5. 如果候选leader在收到的投票中占了多数，则再等待finalizeWait时钟，看是否有人修改leader的候选，如果修改了则把投票放到recvqueue中再从新循环

2. OBSERVING：如果对方是一个观察者，由于它没有投票权，则无视它

3. FOLLOWING或LEADING：
   1. 如果对方和自己再一个时钟周期，说明对方已经完成选举，如果对方说它是leader，那我们就把它作为leader，否则就要比较下对方选举的leader在自己这里是否占有多数，并且选举的leader确认了愿意当leader，如果都通过了，就把这个候选作为自己的leader
   2. 如果对方和自己不再一个时钟周期，说明自己挂掉后又恢复起来，这个时候把别人的投票收集到一个单独的集合outofelection（从名字可以看出这个集合不是用在选举判断），如果对方的投票在outofelection中占有大多数，并且leader也确认了自己愿意做leader，这个时候更新自己的选举周期logicalclock，并修改自己的状态为FOLLOWING或LEADING

 Leader执行流程

QuromPeer线程

1. Leader选举完成之后，Peer确认了自己是Leader的身份，在QuromPeer的主线程中执行Leader的逻辑

2. 创建Leader对象，并创建Server绑定在QuorumAddress上，用于和其他Follower之间相互通信

3. 调用Leader::lead函数，执行Leader的真正的逻辑
   1. 调用ZooKeeperServer::loadData，从磁盘中恢复数据和session列表
   2. 启用新的epoch，zookeeper中的zxid是64位，用于唯一标示一个操作，zxid的高32位是epoch，每次Leader切换加1，低32位为序列号，每次操作加1
   3. 启动绑定在QuorumAddress上的Server，为每个Follower的连接建立一个LearnerHandler，用于和Follower做交互，这里的逻辑另外单独论述
   4. 向所有的Follower发送一个NEWLEADER包，宣告自己额Leader身份，并在initLimit时间内等待大多数的Follower完成和Leader的同步，并发送ACK包，表示Follower已经和Leader完成同步并可以对外提供服务
   5. 这时Leader和Client之间的交互在cnxnFactory的Server中，Leader和Follower之间的交互在LearnerHandler所属的线程中
   6. 然后调用Leader::lead函数的QuromPeer线程在每个tickTime中都会发送2个ping消息给其他的follower，follower在接收到ping消息后会回复一个ping消息，并附带上follower的session tracker里的所有session信息，leader收到follower的ping消息后，根据传回的session信息更新自己的session信息

LearnerHandler线程

1. LearnerHandler主要是处理Leader和Follower之间的交互，和每个Follower连接会维持一个长连接，并有一个单独的LearnerHandler线程和一个Follower进行交互
2. 当Follower和Leader建立连接后，会先发一个FOLLOWERINFO包，包含了follower的server id和最近的一个zxid，即peerLastZxid

3. 根据peerLastZxid来判断如何与Follower进行同步
   1. 如果peerLastZxid大于leader的最新的zxid，则给follower发送trunc包，让follower删掉多出来的事务，一般来说这种情况比较少
   2. 如果peerLastZxid小于leader的最新的zxid，则给follower发送diff包，让follower补齐和leader之间的差距

同步时发送包的顺序如下：

• NEWLEADER（同步发送）

• DIFF（同步发送）

以下包的发送在一个线程中异步发送

• 循环发送写入磁盘的txn和commit包
• 循环发送已经commit但还未写入磁盘的toBeApplied数组的txn和commit包
• 循环发送已经提出proposal但还未commit的outstandingProposals数组中的txn，注意这里没有发送commit包

3. 为了和follower做快速的同步，leader会在内存中缓存一部分最近的事务，即minCommittedLog和maxCommittedLog之间的事务，如果peerLastZxid比minCommittedLog还小的话，leader就给follower发送一个snap包，把当前leader的镜像发给follower

4. 同步等待第一个回复的ACK包，然后计算同步超时tickTime*syncLimit，同步的后续的ACK包在下面的循环中处理
5. 循环处理和follower之间交互的包
   1. ACK包：调用leader.processAck方法，processAck函数的执行逻辑如下：
      1. 如果Ack包的zxid小于Leader的lastCommitted，则忽略
      2. 根据ack包的zxid，在outstandingProposals中找出对应的proposal
      3. 将ack包对应的follower的sid加入proposal的ackset，如果ackset中超过大多数，则表示这个proposal可以commit
      4. 从outstandingProposals中删除这个proposal，并把这个proposal加入到已经可以commit的toBeApplied数组中
      5. 向follower发送commit包，通知follower将proposal提交
   2. PING包：用于和follower同步session信息
   3. REQUEST包：follower转发过来的修改状态的请求，调用ZooKeeperServer::submitRequest方法，这个方法后面单独论述

 

NIOServerCnxn::Factory线程

1. 该线程主要负责server和client的交互

2. 该server是基于select的，当有客户端连接server时，会调用doIO逻辑，这里会把socket上的数据读取出来解析并处理（readPayload函数），并把需要写出的outgoingBuffers写入socket

3. 如果是刚连接上，则调用readConnectRequest，这里会调用

submitRequest(cnxn,sessionId, OpCode.createSession, 0, to, null);

实际是发起一个创建session的请求

4. 如果不是第一次连接，则调用readRequest函数，这里会从socket上读出Request数据，然后调用submitRequest

 

我们可以看到来自client的请求和来自其他server的请求都会调用submitRequest函数，这个函数会调用server上的RequestProcessor链，server实现的是责任链模式，每个请求都会经过责任链里所有RequestProcessor的处理

 

对于Leader来说，LeaderZooKeeperServer::setupRequestProcessors设置了Leader用到的责任链，按从前到后的顺序如下：

• PrepRequestProcessor：创建和修改状态的Request关联的header和txn

• ProposalRequestProcessor：将写请求发送proposal到所有的follower

• SyncRequestProcessor：将发出去的proposal批量写入磁盘

• AckRequestProcessor：当proposal真正写入了磁盘后，向本机发送ack包

• CommitProcessor：匹配本地submitted的请求和收到的committed的请求
• ToBeAppliedRequestProcessor：把写入到磁盘的proposal从toBeApplied中删除

• finalProcessor：把commit的proposal写入到本机的内存状态中