ZAB协议
ZAB协议:如何处理读写请求
你应该有这样的体会,如果你想了解一个网络服务,执行的第一个功能肯定是写操作,然后才会执行读操作。比如,你要了解ZooKeeper,那么肯定会在zkClient.sh命令行中执行写操作(比如create /geekbang 123)写入数据,然后再执行读操作(比如get /geekbang)查询数据。这样一来,你才会直观地理解ZooKeeper的使用方法。
在我看来,任何网络服务最重要的功能就是处理读写请求,因为我们访问网络服务的本质就是执行读写操作,ZooKeeper也不例外,对ZooKeeper而言,这些功能更重要,因为如何处理写请求关乎着操作的顺序性,会影响节点的创建;而如何处理读请求关乎着一致性,也影响着客户端是否会读到旧数据。
接下来,会从ZooKeeper系统的角度全面地分析整个读写请求的流程,帮助你更加全面、透彻地理解读写请求背后的原理。
我们都知道,在ZooKeeper中,写请求必须在领导者上处理,如果跟随者接收到写请求,则需要将写请求转发给领导者,当写请求对应的提案被复制到大多数节点上时,领导者会提交提案,并通知跟随者提交提案。而读请求可以在任何节点上处理,也就是说,ZooKeeper实现的是最终一致性。
所以,理解了如何处理读写请求,不仅能理解读写这个最重要功能的核心原理,还能更好地理解ZooKeeper的性能和一致性。这样在实际场景中安装部署ZooKeeper的时候,就能游刃有余地做资源规划了。比如,如果度请求比较多,可以增加节点,如配置5节点集群,而不是常见的3节点集群。
ZooKeeper处理读写请求的原理。
其实前面已经演示"如何实现操作顺序性"时旧已经介绍了ZooKeeper处理读写请求的原理。这里不再赘述,只在前面的基础上补充几点。
首先,在ZooKeeper中,与领导者"失联"的节点是不能处理读写请求的。比如,如果一个跟随者与领导者的连接发生了读超时,那么它会将自己的状态设置为LOOKING,那么此时它既不能转发写请求给领导者处理,也不能处理读请求,只有当它"找到"领导者后,才能处理读写请求.
例子
- 举个例子,某集群发生分区故障,节点C与节点A(领导者)、节点B断联,那么节点C将设置自己的状态为LOOKING,此时节点C既不能执行读操作,也不能执行写操作。如图所示,
其次,当大多数节点进入广播阶段后,领导者才能提交提案,因为提案提交需要来自大多数节点的确认。最后写请求只能在领导者节点上处理,所以ZooKeeper集群写性能约等于单机。而读请求可以在所有的节点上处理,所以,读性能是水平扩展的。也就是说,你可以通过分集群的方式来突破写性能的限制,并通过增加更加节点来扩展集群的读性能。
ZooKeeper处理读写请求的代码实现
ZooKeeper处理读写请求的具体流程分析如下。
如何实现写操作
在ZooKeeper代码中,处理写请求的核心流程如图所示。这里我用跟随者接收到写请求的情况演示一下。
- 1.跟随者在FollowerRequestProcessor.processRequest()中接收到写请求。具体来说,写请求是系统在ZooKeeperServer.submitRequestNow()中发给跟随者的,如代码所示
firstProcessor.processRequest(si)
而firstProcessor是在FollowerZooKeeperServer.setupRequestProcessors()中创建的,如代码所示
protected void setupRequestProcessors() {
// 创建finalProcessor,提交提案或响应查询
RequestProcessor finalProcessor = new FinalRequestProcessor(this);
// 创建commitProcessor,处理提案提交或读请求
comitProcessor = new CommitProcessor(finalProcessor, Long.toString(getServerId()), true, getZooKeeperServerListener());
commitProcessor.start();
// 创建firstProcessor,接收发给跟随者的请求
firstProcessor = new FollowerRequestProcessor(this,commitProcessor);((FollowerRequest)firstProcessor).start();
// 创建syncProcessor,将提案持久化存储,并返回确认响应给领导者
syncProcessor = new SyncRequestProcessor(this, new SendAckRequestProcessor(getFollower()));
syncProcessor.start();
}
需要注意的是,跟随者节点和领导者节点的firstProcessor是不同的,这样firstProcessor在ZooKeeperServer.submitRequestNow()中被调用时,就分别进入了跟随者和领导者的代码流程。另外,setupRequestProcessors()创建了两条处理链,如图所示。
- 2.跟随者在FollowerRequestProcessor.run()中将写请求转发给领导者,如代码所示
// 调用learner.request() 将请求发送给领导者
zks.getFollower().request(request);
- 3.领导者在LeaderRequestProcessor.processRequest()中接收写请求,并最终调用pRequest()创建事务(也就是提案)并持久化存储,如代码所示
// 创建事务
pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request, true);
......
// 分配事务标识符
request.zxid = zks.getZxid();
// 调用ProposalRequestProcessor.processRequest()处理写请求,并将事务持久化存储
nextProcessor.processRequest(request);
需要注意的是,写请求也是在ZooKeeperServer.submitRequestNow()中发给领导者的,如代码所示
firstProcessor.processRequest(si)
而firstProcessor是在LeaderZooKeeperServer.setupRequestProcessors()中创建的,如代码所所示:
protected void setupRequestProcessors() {
// 创建finalProcessor,最终提交提案和响应查询
RequestProcessor finalProcessor = new FinalRequestProcessor(this);
// 创建toBeAppliedProcessor,存储可提交的提案,并在提交提案后从toBeApplied队列移除已提交的提案
RequestProcessor toBeAppliedProcessor = new Leader.ToBeAppliedRequestProcessor(finalProcessor, getLeader());
// 创建commitProcessor,处理提案提交或读请求
commitProcessor = new COmmitProcessor(toBeAppliedProcessor, Long.toString(getServerId()),false, getZooKeeperServerListener());
commitProcessor.start();
// 创建proposalProcessor,按照顺序广播提案给跟随者
ProposalRequestProcessor proposalProcessor = new ProposalRequestProcessor(this, commitProcessor);
proposalProcessor.initialize();
// 创建preRequestProcessor,根据请求创建提案
preRequestProcessor = new PreRequestProcessor(this, proposalProcessor);
preRequestProcessor.start();
// 创建firstProcessor,接收发给领导者的请求
firstProcessor = new LeaderRequestProcess(this, preRequestProcessor);
...
}
需要注意的是,与跟随者类似,setupRequestProcessor()也为领导者创建了两条处理链(其中处理链2是在创建proposalRequestProcessor时创建的),如图所示.其中,处理链1是核心处理链,最终实现写请求处理(创建提案、广播提案、提交提案)和读请求对应的数据响应。处理链2实现提案持久化存储,并返回确认响应给领导者自己
- 4.领导者在ProposalRequestProcessor.processRequest()中调用propose()将提案广播给集群所有节点,如代码所示:
zks.getLeader().propose(request);
- 5.跟随者在Follower.processPacket()中接收到提案,持久化存储,并返回确认响应给领导者,如代码所示
fzk.logRequest(hdr, txn, digest);
- 6.领导者在接收到大多数节点的确认响应(Leader.processAck())后,最终在CommitProcessor.tryToCommit()提交提案,并广播COMMIT消息给跟随者,如代码所示
// 通知跟随者提交
commit(zxid);
// 自己提交
zk.commitProcessor.commit(p.request);
- 7.跟随者接收到COMMIT消息后,在FollowerZooKeeperServer.commit()中提交提案,如果最初的写请求是自己接收到的,则返回成功响应给客户端,如代码所示
// 必须顺序提交
long firstElementZxid = pendingTxns.element().zxid;
if (firstElementZxid != zxid) {
LOG.error("Commiting zxid 0x" + Long.toHexString(zxid) + "but next pending txn 0x" + Long.toHexString(firstElementZxid));
ServiceUtils.requestSystemExit(ExitCode.UNMATCHED_TXN_COMMIT.getValue());
}// 将准备提交的提案从pendingTxns队列移除
Request request = pendingTxns.remove();
request.logLatency(ServiceMetrics.getMetrics().COMMIT_PROPAGRATION_LATENCY);
// 最终调用FinalRequestProcessor.processRequest() 提交提案,如果最初的写请求是自己接收到的,则返回成功响应给客户端
commitProcessor.commit(request);
这样,ZooKeeper就完成了写请求的处理。需要特别注意的是,在分布式系统中,消息或者核心消息的持久化存储很关键,也很重要,因为这是保证集群稳定运行的关键。当然数据写入最终还是为了后续的数据读取,那么ZooKeeper是如何实现读操作的呢?