zookeeper代码解析

问题导读
1、如何在zoo.cfg中实现集群？
2、total order broadcast特性有哪些？
3、QuorumPeer根据什么状态之一创建对象？

ZooKeeper是近期比较热门的一个类Paxos实现。也是一个逐渐得到广泛应用的开源的分布式锁服务实现。被认为是Chubby的开源版，虽然具体实现有很多差异。ZooKeeper概要的介绍可以看官方文档：http://hadoop.apache.org/zookeeper 这里我们重点来看下它的内部实现。
ZooKeeper集群中的每个server都要知道其他成员，通过在配置文件zoo.cfg中作如下配置实现：

tickTime=2000
dataDir=/var/zookeeper/
clientPort=2181
initLimit=5
syncLimit=2
server.1=zoo1:2888:3888
server.2=zoo2:2888:3888
server.3=zoo3:2888:3888
复制代码

其中第一个端口（端口1）用来做运行期间server间的通信，第二个端口（端口2）用来做leader election，另外还有一个端口（端口0）接收客户端请求。每个机器的这份文件都可以相同。那么一台机器怎样确定自己是谁呢？通过dataDir目录下的myid文本文件确定。myid文件只包含一个数字，内容就是所在Server的ID：QuorumPeer.myid。

构成Zookeeper集群的所有节点，称作ensemble。增加ensemble中的投票节点数，可以提高Zookeeper的QPS，但是写入的效率会下降，因为每个写入操作要在至少过半的投票节点达成一致。投票节点的增加。为了解决这个问题，Zookeeper引入了一个新的节点类型：Observer，与follower相比，只做投票之外的事情，不参与一致性协议的达成。这样通过增加Observer节点即可以提高读吞吐量，又不影响写入的性能，只是可靠性仍然与原先相同，由投票节点的个数决定。

ZooKeeper的启动类是org.apache.zookeeper.server.quorum.QuorumPeerMain 启动时传入配置文件zoo.cfg的路径。QuorumPeerMain解析各项配置，如果发现server列表只有一个，那么直接通过ZooKeeperServerMain来启动单机版的Server；如果有多个，那么读取server列表和myid文件，启动QuorumPeer线程（QuorumPeer继承了Thread，以下直接以线程类作为线程名称）。每个QuorumPeer线程启动之前都会先启动一个cnxnFactory线程，作为nio server接受客户端请求。QuorumPeer线程启动之后，首先做leader election。一个QuorumPeer代表一个ZooKeeper节点，或者说一个ZooKeeper进程。QuorumPeer共有4个状态：LOOKING, FOLLOWING, LEADING, OBSERVING;启动时初始状态是LOOKING，表示正在寻找确定leader中。Leader election的默认算法是基于TCP实现的fast Paxos算法，由FastLeaderElection实现。Leader election的具体实现在淘宝核心系统团队已经有一篇Blog分享过了：http://rdc.taobao.com/blog/cs/?p=162 这里不再赘述。QuorumPeer线程调用FastLeaderElection.lookForLeader选择leader，该方法会在确定leader之后改变QuorumPeer的状态为LEADING, FOLLOWING 或 OBSERVING。QuorumPeer根据Leader election确定的这3个状态之一对应创建LeaderZooKeeperServer、FollowerZooKeeperServer、ObserverZooKeeperServer和Leader、Follower、Observer对象，并调用各自的lead、followLeader、observeLeader方法，如下图所示:

下面我们分别以leader 和 follower的角度看下server接下来的行为。在这之前需要对ZookeeperServer的处理器链有一个了解。单机版Server、Leader、Follower、Observer分别对应ZooKeeperServer、LeaderZooKeeperServer、FollowerZooKeeperServer、ObserverZooKeeperServer。4种Server共享Processor处理器，各自将某几个Processor按顺序组合为一个Processor链。在每个Server中请求总是从第一个Processor开始处理，处理完交给下一个，直到走完整个Processor链。4种Server的Processor链组合如下图所示：

zookeeper-3.3.1-processor-chain

Leader
当QuorumPeer线程确定自己是Leader后，调用Leader对象的lead方法。lead方法首先通过LeaderZooKeeperServer的setupRequestProcessors方法初始化处理器链，启动3个processors线程：
1. PrepRequestProcessor线程。该线程消费请求队列submittedRequests，开始实施一致性算法。submittedRequests有两个来源，一是接入的客户端直接提交，提交的请求既包括写请求，也包括一些查询请求；另一个是由Follower转发，转发内容只包括写请求和同步请求。PrepRequestProcessor收到submittedRequest后，将请求转发给CommitProcessor线程和SyncRequestProcessor线程的输入队列；对于其中的写请求，向所有follower发送PROPOSAL消息（异步发送）。

2. CommitProcessor线程。该线程主要消费两个队列queuedRequests和committedRequests。queuedRequests保存PrepRequestProcessor线程下发的submittedRequest消息。committedRequests保存Proposal通过后，LearnerHanlder线程（后文会有说明）发来的提交请求。CommitProcessor在这里做了如下处理：对于queuedRequests中客户端的查询request，直接返回本地数据；对于客户端提交的或follower转发来的写请求，作为一个pendingRequest等待相应的表决结果返回committedRequest到committedRequests队列。对于队列中到来的每一个committedRequest,如果当前有pendingRequest等待，并且其sessionId，zxid和这个请求匹配，则处理pendingRequest（如果原始请求发自客户端，pendingRequest会携带客户端连接对象，从而能够发送响应给客户端），否则直接处理committedRequest（这种情况对应Follower中的CommitProcessor直接接收到了commit消息）。处理的过程是记录committedLog，变更本地数据。如果请求从客户端来，发送响应给客户端。那么如果一个pendingRequest始终等不到对应的committedRequest到来呢？答案是会一直等待，从而会阻止之后所有queuedRequest请求的处理！开始看到这里以为是个bug，后来想想，如果发生这种情况，已经说明Zookeeper的voter节点超过半数Fault了（不管是消息丢失还是宕机）。这时整个Zookeeper服务只能是不可用了。否则只要过半的voter节点可用，一定会有相应的committedRequest返回。同时这里也保证了写请求按到达顺序生效。

3. SyncRequestProcessor线程。该线程负责将submittedRequest记录到Log。ZooKeeper使用一个简单的内存数据库ZKDatabase来处理日志、session信息和datatree（znode树，类似文件系统结构，用来组织存放实际数据。与文件系统不同的是目录也可以有数据）日志采用1000条批量flush到日志文件，满一定条数起单独线程生成snap文件。记录完日志后直接发送ACK消息给Leader对象—作为一个投票者投出自己的一票
在启动了这3个Processor线程后，Leader对象的lead方法会启动一个LearnerCnxAcceptor线程。LearnerCnxAcceptor线程监听端口1，对接入的每一个Follower连接，启动LearnerHandler线程。启动了LearnerCnxAcceptor线程后，主要的活动交由每个LearnerHandler线程执行。lead方法（QuorumPeer线程）本身进入一个无限循环，向每个Follower定时发送PING消息，当检查到（包括自己）超过半数voter没有响应时，停止整个server。下图是Leader节点的整个线程和队列交互试图，图中颜色和文字相同的为同一个Queue：

下面重点来看下LearnerHandler线程。这个线程处理所有Learner（包括Follower和Observer）的交互逻辑。从Learner发来的消息有以下几种：
1. ACK消息。这是Follower对PROPOSAL消息的响应。Leader收到这个消息后，判断对应的PROPOSAL如果有过半的voter通过，则发送commit请求到CommitProcessor线程的CommittedRequest队列，并且发送Commit消息给所有Follower，发送INFORM消息给所有Observer（告诉这个Proposal通过了）。
2. REQUEST消息。这是Follower转发来的写请求，或者同步请求。转交给PrepRequestProcessor线程处理（放入其submittedRequests队列）
3. PING消息。Learner的心跳消息
4. REVALIDATE消息。用来延长session有效时间

Follower
当QuorumPeer线程确定自己是Follower后，调用Follower对象的followLeader方法。follower通过发送FOLLOWERINFO消息向Leader注册自己，这个消息携带follower自己可以看到的最后一个更新的zxid：peerLastZxid，Leader根据peerLastZxid确定应该向这个follower发送什么样的同步指令，例如是只更新某几天记录，还是发送整个snap。然后发送NEWLEADER消息作为响应，这个消息会携带相应的信息告诉follow怎样同步以和Leader的状态(当前数据)保持一致。当同步完成之后，follower启动Processor线程，进入消息循环。
Follower包含如下几个线程：

1. Follower的QuorumPeer线程：与Leader同步，启动Processor线程和接收客户端请求的nio server线程，循环处理Leader发来的消息
2. NIOServerCnxn.Factory线程：处理客户端请求，认证，维护session时效，转发客户端消息到FollowerRequestProcessor
3. FollowerRequestProcessor线程：处理客户端请求，转发给CommitProcessor线程（放入其队列）。如果是写请求，发送REQUEST消息给Leader。
4. CommitProcessor线程：与Leader中的CommitProcessor线程完全相同—同一个类，同一份代码。只是next Processor挂的直接是FinalRequestProcessor
5. SyncRequestProcessor线程：与Leader中的SyncRequestProcessor线程完全相同—同一个类，只是next Processor挂的是SendAckRequestProcessor，SendAckRequestProcessor负责发送ACK给Leader
下面是Follow节点的线程视图，包括了线程、消息和队列的交互

zookeeper-3.3.1-thread-follower

Follower的消息循环处理如下几种来自Leader的消息
1. PING 心跳消息，返回PING消息给Leader
2. PROPOSAL消息：放入pendingTxns队列，然后转发给SyncRequestProcessor线程
3. COMMIT消息：取出pendingTxns队列中的第一个消息，与这个commit消息比较，如果两者zxid相同，提交给commitProcessor线程处理；如果zxid不同，说明之间有消息丢失，本节点的数据已经不一致了。直接退出server！等下次重启时，再通过和Leader的交互完成数据的同步。
4. UPTODATE消息：Follower开始接入时，在Leader发送完对Follower的同步指令之后，发送这个消息，表示follower可以提供服务了。follower处理该消息时，表名同步已经完成，将当前日志写入snap文件持久化。
5. REVALIDATE消息：根据Leader的REVALIDATE结果，关闭待revalidate的session还是允许其接受消息
6. SYNC消息：返回SYNC结果到客户端。这个消息最初由客户端发起，用来强制得到最新的更新。对应于Paxos协议中的慢速读。

Observer
与Follower类似，只是不参与投票。只进行学习（同步），处理客户端查询请求，转发写请求
消息视图
从上文线程视图中消息、队列的交互处理过程，我们可以提取出ZooKeeper的消息协议。下图总结了ZooKeeper的消息流向：

zookeeper-3.3.1-message

从图中可以看出。对于写操作来说，Server间达成一致的过程其实是一个类似两阶段提交协议的过程：先给所有Voter发送Proposal消息，接收到过半的ACK后就认为更新可以通过，给所有essemble成员发送Commit消息（对Observer发送的其实是INFORM消息）：

zookeeper-3.3.1-message-flow1

在2N+1个voter的集群中，小于等于N个Voter失败的情况下，仍然能处理下去：

zookeeper-3.3.1-message-flow2

Zookeeper最特别的一点是，Leader在发送PROPOSAL消息之前，和Follower接收到PROPOSAL消息之后，都会立即将消息记录到日志中。这样在收到过半的ACK之后，既可以确认消息已经在过半的server中保存过了。即使之后的Commit消息发送失败，也在事实上通过了消息。丢失commit消息的follower会在下一个事务中发现这一点，并自动退出。通过重启来重新取得一致性。（这里似乎没有看到自动重启的机制。。。）
在Zookeeper的官方文档中，提到了Zookeeper的Atomic Broadcast特性。Atomic Broadcast特性即total order broadcast特性：

Reliable delivery
如果一个消息m被某一个server递交，这个消息最终将会被所有server递交。

Total order
如果在某一个server上，消息a在消息b之前递交，那么在所有的server上，消息a都会在消息b之前递交。如果a和b是已递交的消息，要么a在b之前递交，要么b在a之前递交。

Causal order
如果消息b在b的发送者递交a之后发送，a一定会在b之前。如果一个发送者发送了b之后再发送c，c一定会在b之后。
通过上述ZooKeeper的代码分析，我们看到，Server间的一致性协议保证了消息的可靠递送（Reliable delivery）；Server内部所有处理器的单线程加FIFO队列处理模式，保证了消息的全局顺序（total order）和因果顺序（causal order）；消息日志的内存化保证了系统的效率。
ZooKeeper的代码整体上来说比较清晰。大的模块划分井然有序，杂而不乱。并且在复杂的消息处理，一致性协议的实现中，通过ZooKeeperServer和RequestProcessor两个体系达到了尽可能多的代码复用。