Paxos算法之旅(四)zookeeper代码解析

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方法,如下图所示:

zookeeper-3.3.1-start

下面我们分别以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:

zookeeper-thread-leader

下面重点来看下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两个体系达到了尽可能多的代码复用。

## 其他模块

待续。。。

个人资料
bjchenli
等级:8
文章:260篇
访问:22.0w
排名: 3
上一篇: BTrace使用简介
下一篇:Makefile知识点整理
猜你感兴趣的圈子:
阿里中间件技术交流圈
标签: follower、leader、消息、quorumpeer、线程、面试题
隐藏