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方法.

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

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。下

下面重点来看下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

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最特别的一点是,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两个体系达到了尽可能多的代码复用。

本文来源于"阿里中间件团队播客",原文发表时间" 2010-11-10"

时间: 2024-10-31 20:10:34

Paxos算法之旅(四)zookeeper代码解析的相关文章

Zookeeper笔记(二)Paxos算法与Zookeeper的工作原理

Zookeeper 分布式服务框架是 Apache Hadoop 的一个子项目, 它主要是用来解决分布式应用中经常遇到的一些数据管理问题,如:统一命名服务.状态同步服务.集群管理.分布式应用配置项的管理等. paxos算法 Zookeeper 采用paxos一致性算法保证了数据的一致性,Paxos算法是一种基于消息传递且具有高度容错特性的一致性算法. 具体的算法不多作介绍,可以查看维基百科Paxos算法. 想要更好的理解Paxos算法,可以关注知乎的这个问题 如何浅显易懂地解说 Paxos 的算

Zookeeper全解析——Paxos作为灵魂

Zookeeper全解析--Paxos作为灵魂 原计划在介绍完ZK Client之后就着手ZK Server的介绍,但是发现ZK Server所包含的内容实在太多,并不是简简单单一篇Blog就能搞定的.于是决定从基础搞起比较好. 那么ZK Server最基础的东西是什么呢?我想应该是Paxos了.所以本文会介绍Paxos以及它在ZK Server中对应的实现. 先说Paxos,它是一个基于消息传递的一致性算法,Leslie Lamport在1990年提出,近几年被广泛应用于分布式计算中,Goog

c语言-编程算法 - 最小的k个数 代码(C)

问题描述 编程算法 - 最小的k个数 代码(C) 请解释一下在c语言中怎样编写在输入的N个数中找到k个最小的数 解决方案 排序吧,再输出前k个数 解决方案二: 遍历,找出MAX,移除MAX,循环K遍 解决方案三: 我觉得你的问题是怎么将输入的数保存下来,你可以先定义一个vector. vector vec;int iNUm = 0;while(cin>>iNum)//需要结束的时候输入ctrl+z;{ vec.push_back(iNum);}//最后对整个vec进行排序,取得最小的值 解决方

如何将aspx页面代码解析成html?

问题描述 如:页面中包含<asp:TextBoxID="TextBox1"runat="server"filed="codeNo"dataType="nvarchar"dataSize="50"></asp:TextBox> 写代码解析成<inputname="TextBox1"type="text"id="TextBox1&q

分布式系列文章——Paxos算法原理与推导

点击我的博客查看原文. Paxos算法在分布式领域具有非常重要的地位.但是Paxos算法有两个比较明显的缺点:1.难以理解 2.工程实现更难. 网上有很多讲解Paxos算法的文章,但是质量参差不齐.看了很多关于Paxos的资料后发现,学习Paxos最好的资料是论文<Paxos Made Simple>,其次是中.英文版维基百科对Paxos的介绍.本文试图带大家一步步揭开Paxos神秘的面纱. Paxos是什么 Paxos算法是基于消息传递且具有高度容错特性的一致性算法,是目前公认的解决分布式一

Java二进制指令代码解析

小注:去年在看<深入解析JVM>书的时候做的一些记录,同时参考了<Java虚拟机规范>.只是对指令的一些列举,加入了一些自己的理解.可以用来查询. Java二进制指令代码解析 Java源码在运行之前都要编译成为字节码格式(如.class文件),然后由ClassLoader将字节码载入运行.在字节码文件中,指令代码只是其中的一部分,里面还记录了字节码文件的编译版本.常量池.访问权限.所有成员变量和成员方法等信息(详见Java字节码格式详解).本文主要简单介绍不同Java指令的功能以及

算法难题设计出java代码或者伪代码,大牛请进。

问题描述 算法难题设计出java代码或者伪代码,大牛请进. 把 1 2 3 4 5 6 7 8 9 放入三个数组里面 数组可以是空的.. 数组里面的数 是有序的 比如 {1 2 3} { 4 5 6 } { 7 8 9 }:{356789},{124},{}能穷举吗.打印出来 解决方案 {123456789},{},{} 可以么,如果是可以的话,那么是非常简单的 解决方案二: 我是一个刚刚学习编程半年的小白,有点思路,可能不准确,抛砖引玉.我觉得这个题的实质,是对1 2 3 4 5 6 7 8

uploadify上传及后台文件合法性验证的代码解析_java

后台上传方法 @RequestMapping(value = "/api_upload", method = RequestMethod.POST) public @ResponseBody String upload(HttpServletRequest request,HttpServletResponse response) { //获取上传路径 String uploadFilePath=ParameterConstants.UPLOAD_FILE_PATH; String s

大数据情况下桶排序算法的运用与C++代码实现示例_C 语言

箱排序的变种.为了区别于上述的箱排序,姑且称它为桶排序(实际上箱排序和桶排序是同义词). 桶排序的思想是把[0,1)划分为n个大小相同的子区间,每一子区间是一个桶.然后将n个记录分配到各个桶中.因为关键字序列是均匀分布在[0,1)上的,所以一般不会有很多个记录落入同一个桶中.由于同一桶中的记录其关键字不尽相同,所以必须采用关键字比较的排序方法(通常用插入排序)对各个桶进行排序,然后依次将各非空桶中的记录连接(收集)起来即可. 注意: 这种排序思想基于以下假设:假设输入的n个关键字序列是随机分布在