版权信息

书名：Zookeeper源码分析

作者：侯宜军

本书由北京图灵文化发展有限公司发行数字版。版权所有，侵权必究。

您购买的图灵电子书仅供您个人使用，未经授权，不得以任何方式复制和传播本书内容。

我们愿意相信读者具有这样的良知和觉悟，与我们共同保护知识产权。

如果购买者有侵权行为，我们可能对该用户实施包括但不限于关闭该帐号等维权措施，并可能追究法律责任。

图灵社区会员 A_Q（961353968@qq.com） 专享 尊重版权

内容摘要

作者简介

1　入门

1.1　简介

1.2　源码

1.3　客户端命令行

1.4　Java API

1.4.1　create

1.4.2　删除

1.4.3　getData

1.4.4　setData

1.4.5　getChildren

1.4.6　Watcher

2　客户端

2.1　ZooKeeper

2.1.1　create

2.1.2　delete

2.1.3　Exists

2.1.4　getData

2.1.5　setData

2.1.6　getChildren

2.2　通信机制

2.3　ClientCnxn

2.3.1　submitRequest

2.3.2　queuePacket

2.3.3　sendPacket

2.3.4　finishPacket

2.4　ClientCnxnSocketNetty

2.4.1　doTransport

2.4.2　ZkClientHandler.messageReceived

2.5　Packet类

2.6　sendThread类

2.6.1　sendPacket

2.6.2　readResponse

2.7　StaticHostProvider

2.7.1　updateServerList

3　投票选举

3.1　QuorumPeer

3.2　QuorumCnxManager

3.2.1　发送线程

3.2.2　接收线程

3.3　Messenger

3.3.1　Notification类

3.3.2　ToSend类

3.3.3　WorkerReceiver

3.3.4　WorkerSender

3.4　快速选举

3.4.1　lookForLeader

4　事务处理

4.1　QuorumPacket

4.2　Learner

4.2.1　request

4.2.2　注册到Leader

4.3　Follower

4.3.1　followLeader

4.3.2　processPacket

4.4　Observer

4.4.1　observeLeader

4.4.2　processPacket

4.5　Leader

4.5.1　响应Follower注册

4.5.2　事务处理

5　服务端

5.1　启动类

5.2　ZooKeeperServer

5.2.1　客户端请求

5.2.2　主要方法

5.2.3　processPacket

5.2.4　processTxn

5.3　调用链

5.3.1　LeaderRequestProcessor

5.3.2　FollowerRequestProcessor

5.3.3　ObserverRequestProcessor

5.3.4　ProposalProcessor

5.3.5　CommitProcessor

5.3.6　SyncRequestProcessor

5.3.7　AckRequestProcessor

5.3.8　SendAckRequestProcessor

5.3.9　toBeAppliedProcessor

5.3.10　PrepRequestProcessor

5.3.11　FinalRequestProcessor

5.4　LeaderZooKeeperServer

5.4.1　调用链

5.4.2　写请求

5.4.3　读请求

5.4.4　submitLearnerRequest

5.5　FollowerZooKeeperServer

5.5.1　调用链

5.5.2　写请求

5.5.3　读请求

5.5.4　logRequest

5.5.5　commit

5.6　ObServerZooKeeperServer

5.6.1　调用链

5.6.2　commitRequest

5.7　NettyServerCnxn

6　Zookeeper数据库

6.1　DataTree

6.1.1　概述

6.1.2　serialize

6.1.3　deserialize

6.1.4　　processTxn

6.2　DataNode

7　Watcher机制

7.1　Watcher接口

7.2　客户端

7.3　服务端

7.4　WatchManager

7.4.1　addWatch方法

7.4.2　triggerWatch方法

7.5　NettyServerCnxn

7.6　ClientCnxn

8　事务持久化

8.1　概述

8.2　接口

8.2.1　TxnLog

8.2.2　SnapShot

8.3　实现类

8.3.1　FileTxnLog

8.3.2　FileSnap

9　命令行

9.1　概述

9.2　bin目录

9.2.1　ZkServer.cmd

9.2.2　ZKCli.cmd

9.3　ZooKeeperMain

9.3.1　Run方法

9.3.2　executeLine

9.3.3　processCmd

9.3.4　processZKCmd

9.4　cli目录

9.4.1　CliCommand

9.4.2　GetCommand

9.4.3　SetCommand

9.4.4　StatCommand

9.4.5　LsCommand

10　配置和应用

10.1　管理端

10.1.1　启动

10.1.2　JettyAdminServer

10.1.3　CommandServlet

10.1.4　Commands

10.2　服务器配置

10.3　ZooKeeper在大数据中的应用

10.3.1　实时配置文件

10.3.2　Kafka中使用

10.3.3　Hadoop使用

10.4　参考

内容摘要

大数据和人工智能时代，涌现了越来越多的开源框架，这些软件框架都有一个基本的共同点，就是大数据量的计算从而需要用到大量的机器集群作为支撑。那么一套分布式运算协调管理的软件就是必要条件之一。而ZooKeeper正是一款广泛应用于各种分布式计算场景的成熟开源框架。

目前很多程序员在从事分布式系统的开发，使用过ZooKeeper。但很大一部分人还停留在知其然不知其所以然的阶段，只是知道怎样使用API编程接口，对其核心技术的实现机制知之甚少，无形中增加了其在分布式应用领域往前精进的难度。

本书并不是一本原理性的介绍分布式系统设计原理方面的书，而是从源码出发，学习世界顶尖程序员的编程思想。本书共分成十个章节。第一章介绍ZooKeeper的入门级使用；第二章介绍ZooKeeper客户端源码的设计思路；第三章到第八章是服务端部分的源码解析，包括服务端的启动、投票选举过程、响应客户端请求机制、写数据一致性机制等内容；第九章主要介绍服务端命令行启动的流程；第十章介绍ZooKeeper的管理和应用场景。

本书的目标人群定位为具有一定经验，对流行开源软件源码有浓厚的学习兴趣，致力于提高自身软件开发能力的的软件工程师。

作者简介

侯宜军，男，南京邮电大学计算机系研究生毕业，先后在电信设计院、摩托罗拉、医疗互联网初创公司等工作过，居住在南京。

具有多年分布式数据采集，数据仓库建模从业经历，对HBase，Storm，Spark，Kafka等大数据技术较熟悉。目前研究方向集中在分布式系统、机器学习框架研究、神经网络等领域。

2015年底与他人共同创立六度服务号中医在线平台，2017年初因个人原因退出创业团队，目前在苏宁云商任职高级技术经理。

1　入门

1.1　简介

Zookeeper 分布式服务框架是Apache Hadoop 的一个子项目，它主要是用来解决分布式应用中经常遇到的一些数据管理问题，如：统一命名服务、状态同步服务、集群管理、分布式应用配置项的管理等。

Zookeeper分成客户端和服务端两部分，服务端通常运行于集群模式，统一为客户端提供分布式数据一致性管理服务，集群可提供水平扩展能力，以及避免单点故障。

ZooKeeper让分布式进程可通过共享的、与标准文件系统类似的分层名字空间相互协调。名字空间由数据寄存器（在ZooKeeper世界中称作znode）构成，这与文件和目录类似。与用于存储设备的典型文件系统不同的是，ZooKeeper的数据往往常驻内存提供服务，提高性能。

Zookeeper还提供了数据动态监测机制，客户端注册监听它关心的目录节点，当目录节点发生变化（数据改变、被删除、子目录节点增加删除）时，客户端会触发注册事件客户端。

ZooKeeper的实现很重视高性能、高可用性，以及严格的顺序访问。高性能意味着可将ZooKeeper用于大的分布式系统。可靠性使之可避免单点失败。严格的顺序访问使得客户端可以实现复杂的同步原语。

服务端通常有多台服务器，这些服务器都通过QuorumPeer进程维持分布式系统间的数据交换，选举领导者（Leader）和追随者（Follower），动态监控集群，一旦发生节点故障则重新启动选举流程，选出新的Leader。每个服务器维护相同的一份全局数据，通过QuorumPeer维持数据之间的同步，这样提供高可靠性和横向扩展性。

1.2　源码

本书以V3.3.0版本为例，总java代码行数约为11万行。

源码地址：https://github.com/apache/zookeeper

文件目录结构，为了方便解析整个ZooKeeper的源码结构，这里只是列出主要的目录和文件：

目录

说明

/

根目录下主要是ZooKeeper客户端相关的文件

/cli

命令行方式的Java类

/server

ZooKeeperServer相关，为ZooKeeper客户端提供服务

/server/command

服务端命令

/server/admin

ZooKeeper服务端Web管理，基于Jetty实现

/server/persistence

事务日志和快照的持久化

/server/quorum

集群间状态维护和切换，实现分布式事务

1.3　客户端命令行

本节从Zookeeper客户端的角色出发，理解Zookeeper的常用操作方法。

客户端提供的操作包括启动、关闭、创建节点、设置节点数据、删除节点、获取节点、获取节点数据等。这里只是取其中的几个命令做为示例。

主要客户端命令表：

ZooKeeper 命令

功能描述

ls

获取子路径列表

create

创建路径和数据

delele

删除路径

get

获取路径的数据

set

设置路径的数据

connect

连接服务端

close

关闭与服务端连接

quit

退出客户端

setacl

设置路径的Acl

getacl

获取路径的Acl

- 启动客户端

zkCli.sh

- 连接服务端

Connect 10.10.27.21

- 创建路径节点

创建一个新的 znode ，使用 create /test “hello” 。这个命令在根目录下创建了一个新的 znode 节点“ test ”以及与它关联的字符串：

- 获取目录

使用 ls 命令来查看当前 ZooKeeper 中所包含的内容，这里显示/目录下有test、zookeeper两个子节点。

- 获取节点数据

运行 get 命令来确认第二步中所创建的 znode，显示znode的数据、创建时间等状态信息。

示例数据：

- 设置节点数据

通过 set 命令来对 zk 所关联的字符串进行设置，设置/test路径的数据为“welcome！”。

- 删除节点

delete  /test

1.4　Java API

ZooKeeper还为客户端提供了Java API编程接口的方式访问操作数据。API接口主要是ZooKeeper类中的方法，下面列举几个常用的客户端操作的Java代码示例。

1.4.1　create

创建ZooKeeper路径节点，参数需提供路径名称和路径关联数据。对应create命令。

示例代码如下：

ZooKeeper zk= new ZooKeeper("127.0.0.1",SESSION_TIME,this);
zk.create("/test","Hello".getBytes(), null, CreateMode.PERSISTENT);

1.4.2　删除

删除路径，对应delete命令。

ZooKeeper zk= new ZooKeeper("127.0.0.1",SESSION_TIME,this);
zk.delete("/test", -1);

1.4.3　getData

获取路径关联数据。对应get命令。

ZooKeeper zk= new ZooKeeper("127.0.0.1",SESSION_TIME,this);
String data  = zk.getData("/test", false,null);

1.4.4　setData

设置节点关联数据。对应set命令。

ZooKeeper zk= new ZooKeeper("127.0.0.1",SESSION_TIME,this);
zk.setData ("/test", "Welcome".getBytes(), -1);

1.4.5　getChildren

获取子节点路径列表。对应ls命令。

ZooKeeper zk= new ZooKeeper("127.0.0.1",SESSION_TIME,this);
List<String> list = zk.getChildren("/", false);

1.4.6　Watcher

设置观察者：

只要在实例化ZooKeeper类时指定Watcher实现即可实现ZooKeeper事件坚挺，Watcher实现只需要实现process接口方法即可。

示例代码：

zk = new ZooKeeper(address, sessionTimeout, new Watcher() {
// 监控所有被触发的事件
public void process(WatchedEvent event) {
System.out.println("触发事件：" + event.getType());
}
});

这段代码监听ZooKeeper所有变化，事件处理代码是打印event类型。

2　客户端

ZooKeeper类是Zookeeper客户端的实现类，使用Zookeeper都是通过这个类来发送命令给Zookeeper服务器的。

ZooKeeper中可以设置Watcher，每个Watcher在Zookeeper的节点状态发生变化时被通知，以执行预先注册的Watcher动作。

Zookeeper包含三种Watcher列表：DataWatcher，ExistWatcher和ChildWatcher。

本章主要是分析ZooKeeper类的实现。

2.1　ZooKeeper

客户端在构造函数阶段创建ClientCnxn 与服务端连接，后续命令都通过ClientCntx发送给服务端。ClientCnxn是客户端与服务端通信的底层接口，它和ClientCnxnSocketNetty一起工作提供网络通信服务。

服务端是ZooKeeperServer类，收到ClientCnxn的Request请求包后调用相应的处理逻辑，返回结果再通过ClientCnxc返回到客户端。

ClientCntx连接时可以同时指定多台服务器地址，根据一定的算法挑选某一个服务器地址进行连接，当某个服务器发生故障无法连接时，会自动连接其他的服务器。实现这一机制的是HostProdiver接口，实现用StaticHostProvider类。

ZooKeeper类的构造函数如下：

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
boolean canBeReadOnly, HostProvider aHostProvider, ZKClientConfig clientConfig)
throws IOException {
if (clientConfig == null) {
clientConfig = new ZKClientConfig();
}
this.clientConfig = clientConfig;
watchManager = defaultWatchManager();
watchManager.defaultWatcher = watcher;
ConnectStringParser connectStringParser = new
ConnectStringParser(connectString);
hostProvider = aHostProvider;
cnxn = new ClientCnxn(connectStringParser.getChrootPath(),
hostProvider, sessionTimeout, this, watchManager,
getClientCnxnSocket(), canBeReadOnly);
cnxn.start();
}

创建了ZooKeeper对象后就可以调用具体的读写数据的方法了，下面列举常见方法的实现机制。

2.1.1　create

客户端调用create方法在Zookeeper集群创建一个Node。返回值为成功创建的路径名称。

实现源码：

public String create(final String path, byte data[], List<ACL> acl,
CreateMode createMode)   throws KeeperException, InterruptedException{
final String clientPath = path;
PathUtils.validatePath(clientPath, createMode.isSequential());
EphemeralType.validateTTL(createMode, -1);
final String serverPath = prependChroot(clientPath);
RequestHeader h = new RequestHeader();
h.setType(createMode.isContainer() ? ZooDefs.OpCode.createContainer : ZooDefs.OpCode.create);
CreateRequest request = new CreateRequest();
CreateResponse response = new CreateResponse();
request.setData(data);
request.setFlags(createMode.toFlag());
request.setPath(serverPath);
if (acl != null && acl.size() == 0) {
throw new KeeperException.InvalidACLException();
}
request.setAcl(acl);
ReplyHeader r = cnxn.submitRequest(h, request, response, null);
if (r.getErr() != 0) {
throw KeeperException.create(
KeeperException.Code.get(r.getErr()),clientPath);
}
if (cnxn.chrootPath == null) {
return response.getPath();
} else {
return response.getPath().substring(cnxn.chrootPath.length());
}
}

create方法根据输入参数构造出CreateRequest包，然后通过submitRequest方法传递给服务端，submitRequest方法将CreateRequest转换成Packet包并调用sendPacket方法将发送包放入队列，等待发送线程发送给服务端。

服务端响应完成后会将返回结果填充到CreateResponse实体中返回给客户端。

2.1.2　delete

有创建节点的操作，自然也有删除节点的操作，这里删除节点同样也提供了同步和异步两种接口方式，具体代码和创建节点类似。

实现源码：

public void delete(final String path, int version, VoidCallback cb,
Object ctx)  {
final String clientPath = path;
PathUtils.validatePath(clientPath);
final String serverPath;
if (clientPath.equals("/")) {
serverPath = clientPath;
} else {
serverPath = prependChroot(clientPath);
}
RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.delete);
DeleteRequest request = new DeleteRequest();
request.setPath(serverPath);
request.setVersion(version);
cnxn.queuePacket(h, new ReplyHeader(), request, null, cb,
clientPath, serverPath, ctx, null);
}

2.1.3　Exists

判断节点是否存在，异步方式，并且在回调函数中触发事件观察者。

public void exists(final String path, Watcher watcher,  StatCallback cb, Object ctx)  {
final String clientPath = path;
PathUtils.validatePath(clientPath);
WatchRegistration wcb = null;
if (watcher != null) {
wcb = new ExistsWatchRegistration(watcher, clientPath);
}
final String serverPath = prependChroot(clientPath);
RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.exists);
ExistsRequest request = new ExistsRequest();
request.setPath(serverPath);
request.setWatch(watcher != null);
SetDataResponse response = new SetDataResponse();
cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
clientPath, serverPath, ctx, wcb);
}

2.1.4　getData

获取节点关联数据，客户端示例代码：

实现源码：

public void getData(final String path, Watcher watcher,            DataCallback cb, Object ctx)  {
final String clientPath = path;
PathUtils.validatePath(clientPath);
WatchRegistration wcb = null;
if (watcher != null) {
wcb = new DataWatchRegistration(watcher, clientPath);
}

final String serverPath = prependChroot(clientPath);
RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.getData);
GetDataRequest request = new GetDataRequest();
request.setPath(serverPath);
request.setWatch(watcher != null);
GetDataResponse response = new GetDataResponse();
cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
clientPath, serverPath, ctx, wcb);
}

2.1.5　setData

设置节点关联数据

实现源码：

public Stat setData(final String path, byte data[], int version)
throws KeeperException, InterruptedException   {
final String clientPath = path;
PathUtils.validatePath(clientPath);
final String serverPath = prependChroot(clientPath);
RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.setData);
SetDataRequest request = new SetDataRequest();
request.setPath(serverPath);
request.setData(data);
request.setVersion(version);
SetDataResponse response = new SetDataResponse();
ReplyHeader r = cnxn.submitRequest(h, request, response, null);
return response.getStat();
}

2.1.6　getChildren

获取子节点路径列表

实现源码：

public List<String> getChildren(final String path, Watcher watcher)
throws KeeperException, InterruptedException   {
final String clientPath = path;
PathUtils.validatePath(clientPath);
WatchRegistration wcb = null;
if (watcher != null) {
wcb = new ChildWatchRegistration(watcher, clientPath);
}
final String serverPath = prependChroot(clientPath);
RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.getChildren);
GetChildrenRequest request = new GetChildrenRequest();
request.setPath(serverPath);
request.setWatch(watcher != null);
GetChildrenResponse response = new GetChildrenResponse();
ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
if (r.getErr() != 0) {
throw KeeperException.create(KeeperException.Code.get(r.getErr()),
clientPath);
}
return response.getChildren();
}

2.2　通信机制

客户端与服务端交互的数据流大致如下：

1. 客户端发起命令请求，这里我们以create命令为例。

2. 构造Packet，调用ClientCnxn的submitRequest方法。ClientCnxn的submitRequest方法调用queuePacket方法将Request放入outgoingQueue队列中，然后调用线程wait方法等待该Request从服务端获得响应为止。

3. ClientCnxnSocketNetty和ClientCnzx共享同一个outgoingQueue，ClientCnzxSocketNetty启动了发送守护进程，当outgoingQueue队列中有Request时，会自动将该Request发送给ZooKeeperServer。同时ClientCnzxSocketNetty启动接收线程实时接收ZooKeeperServer的返回数据，返回数据触发ClientCnzxSocketNetty中启动的ZKClientHandler的MessageReceived事件。

4. 在MessageReceived事件中调用ClientCnxn中的SendThread的readResponse方法。

5. readResponse方法中最后调用finishPacket方法唤醒在该请求上wait的2）中的submitRequest方法。使得submitRequest方法返回到客户端。

6. 客户端获得返回的命令执行结果。

2.3　ClientCnxn

客户端通信要用到ClientCnxn，客户端的命令都是通过该类来与Zookeeper集群通信，发送命令请求并获取命令响应。

Clientcnxn于ZooKeeper都位于Zookeeper的根目录下，处于同一级别。

Clientcnxn将客户端请求加入发送队列，等待sendThread发送。eventThread负责处理Server返回的WatchedEvent，回调注册的客户端事件接口处理函数。

属性名称

说明

outgoingQueue

等待发送请求队列

ZooKeeper

指向客户端

ClientWatchManager

Watcher事件管理

sendThread

发送线程

eventThread

事件处理线程

hostProvider

可以连接的ZK服务器列表

readResponse

当ClientCnzxSocketNetty收到服务端响应包时，通过ZKClientHandler的MessageReceived事件来调用readResponse方法，将响应消息填充到Request请求包的response字段上。

2.3.1　submitRequest

提交客户端请求到服务端。

ReplyHeader r = new ReplyHeader();
Packet packet = queuePacket(h, r, request, response, null, null, null,  null, watchRegistration, watchDeregistration);
synchronized (packet) {
while (!packet.finished) {
packet.wait();
}
}
return r;

2.3.2　queuePacket

将发送包放入发送队列outgoingQueue中，等待发送线程发送给服务端。

public Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
Record response, AsyncCallback cb, String clientPath,
String serverPath, Object ctx, WatchRegistration atchRegistration,
WatchDeregistration watchDeregistration) {
Packet packet = null;
packet = new Packet(h, r, request, response, watchRegistration);        packet.cb = cb;
packet.ctx = ctx;
packet.clientPath = clientPath;
packet.serverPath = serverPath;
packet.watchDeregistration = watchDeregistration;

synchronized (state) {
if (!state.isAlive() || closing) {
conLossPacket(packet);
} else {
if (h.getType() == OpCode.closeSession) {
closing = true;
}
outgoingQueue.add(packet);
}
}
sendThread.getClientCnxnSocket().packetAdded();
return packet;
}

最后通过SendThread发送请求到服务端。

2.3.3　sendPacket

客户端调用SendThread的sendPacket方法发送数据。

public void sendPacket(Record request, Record response, AsyncCallback cb, int opCode)   throws IOException {
int xid = getXid();
RequestHeader h = new RequestHeader();
h.setXid(xid);
h.setType(opCode);
ReplyHeader r = new ReplyHeader();
r.setXid(xid);
Packet p = new Packet(h, r, request, response, null, false);
p.cb = cb;
sendThread.sendPacket(p);
}

2.3.4　finishPacket

接收到服务端的响应，唤醒等待响应的客户端线程。

如果客户端请求指定了Watcher，则同时生成WatchedEvent事件并放入事件队列，等待eventThread线程处理。

客户端请求在写操作的情况下，发送完命令给服务端后会阻塞等待服务端响应，当ClientCntx收到服务端响应后调用finishPacket方法，唤醒阻塞在该Packet的客户端，客户端至此收到响应。

代码片段：

private void finishPacket(Packet p) {
if (p.cb == null) {
synchronized (p) {
p.finished = true;
p.notifyAll();
}
} else {
p.finished = true;
eventThread.queuePacket(p);
}
}

2.4　ClientCnxnSocketNetty

客户端Socket的Netty实现。Netty是一款高性能的Socket框架，广泛应用于各种分布式系统中。

2.4.1　doTransport

发送过程，在待发送队列（outgoingQueue）中取出Request发送给ZooKeeperServer。当ClientCnxn准备发送数据时会调用该方法。

主要代码：

Packet head = null;
if ((head = outgoingQueue.poll(waitTimeOut, TimeUnit.MILLISECONDS)) == null) {
return;
}
if (head != null) {
doWrite(pendingQueue, head, cnxn);
}

doWrite方法会将发送队列中的Request发送给服务端的Socket。

2.4.2　ZkClientHandler.messageReceived

接收事件。当ClientCnzxSocketNetty接收到服务端的返回数据时，会触发该事件，最后调用sendThread.readResponse方法通知到ClientCnzx。

主要代码：

public void messageReceived(ChannelHandlerContext ctx,
MessageEvent e) throws Exception {
updateNow();
ChannelBuffer buf = (ChannelBuffer) e.getMessage();
while (buf.readable()) {
if (incomingBuffer.remaining() > buf.readableBytes()) {
int newLimit = incomingBuffer.position()+ buf.readableBytes();
incomingBuffer.limit(newLimit);
}
buf.readBytes(incomingBuffer);
incomingBuffer.limit(incomingBuffer.capacity());
if (!incomingBuffer.hasRemaining()) {
incomingBuffer.flip();
if (incomingBuffer == lenBuffer) {
recvCount++;
readLength();
} else if (!initialized) {
readConnectResult();
lenBuffer.clear();
incomingBuffer = lenBuffer;
initialized = true;
updateLastHeard();
} else {
sendThread.readResponse(incomingBuffer);
lenBuffer.clear();
incomingBuffer = lenBuffer;
updateLastHeard();
}
}
}
wakeupCnxn();
}

2.5　Packet类

请求响应包，Clientcnxn发送请求到服务端都是将请求数据和响应数据以及连接信息封装到Packet类中进行处理，包括线程阻塞等待响应结果的到来也是封装在该类中。

Packet的主要属性定义如下：

static class Packet {
RequestHeader requestHeader;
ReplyHeader replyHeader;
Record request;
Record response;
ByteBuffer bb;
String clientPath;
String serverPath;
boolean finished;
AsyncCallback cb;
Object ctx;
WatchRegistration watchRegistration;
public boolean readOnly;
WatchDeregistration watchDeregistration;
}

2.6　sendThread类

客户端用于与服务端收发数据的主线程，当Netty获得服务端响应时最终会触发readResponse方法被调用，读取服务端响应数据。

sendPacket可被用来发送Request请求给服务端，将Request放入Netty的发送队列中等待发送。

2.6.1　sendPacket

发送接口，

public void sendPacket(Packet p) throws IOException {
clientCnxnSocket.sendPacket(p);
}

2.6.2　readResponse

ClientCnzxSocketNetty收到服务端响应后触发ZKClientHandler的messageReceived事件，在该事件处理逻辑中调用sendThread的readResponse方法获取服务端响应。

如果服务端响应的是WatchedEvent事件，则将事件放入eventThread中等候调度执行事件方法。

如果服务端响应的是客户端命令结果，则组装到Packet到返回给CientCnxn的finishPacket方法，由finishPacket方法再返回给客户端。前面提过，finishPacket方法填充在ClientCnzx的submitRequest方法执行时阻塞的Packet包中的Response数据结构，并解除阻塞，使得ClientCnzx的submitRequest方法得以执行结束，顺利返回到ZooKeeper的create方法调用，此时create命令执行完毕。

完整的读取服务端响应的代码如下：

void readResponse(ByteBuffer incomingBuffer) throws IOException {
ByteBufferInputStream bbis = new ByteBufferInputStream(
incomingBuffer);
BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
ReplyHeader replyHdr = new ReplyHeader();

replyHdr.deserialize(bbia, "header");
if (replyHdr.getXid() == -2) {
// -2 is the xid for pings
if (LOG.isDebugEnabled()) {
LOG.debug("Got ping response for sessionid: 0x"
+ Long.toHexString(sessionId)
+ " after "
+ ((System.nanoTime() - lastPingSentNs) / 1000000)
+ "ms");
}
return;
}
if (replyHdr.getXid() == -4) {
// -4 is the xid for AuthPacket
if(replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) {
state = States.AUTH_FAILED;
eventThread.queueEvent( new
WatchedEvent(Watcher.Event.EventType.None,
Watcher.Event.KeeperState.AuthFailed, null) );
}
if (LOG.isDebugEnabled()) {
LOG.debug("Got auth sessionid:0x"
+ Long.toHexString(sessionId));
}
return;
}
if (replyHdr.getXid() == -1) {
// -1 means notification
WatcherEvent event = new WatcherEvent();
event.deserialize(bbia, "response");
// convert from a server path to a client path
if (chrootPath != null) {
String serverPath = event.getPath();
if(serverPath.compareTo(chrootPath)==0)
event.setPath("/");
else if (serverPath.length() > chrootPath.length())
event.setPath(serverPath.substring(chrootPath.length()));
else {
LOG.warn("Got server path " + event.getPath()
+ " which is too short for chroot path "
+ chrootPath);
}
}
WatchedEvent we = new WatchedEvent(event);
if (LOG.isDebugEnabled()) {
LOG.debug("Got " + we + " for sessionid 0x"
+ Long.toHexString(sessionId));
}
eventThread.queueEvent( we );
return;
}
Packet packet;
synchronized (pendingQueue) {
if (pendingQueue.size() == 0) {
throw new IOException("Nothing in the queue, but got "
+ replyHdr.getXid());
}
packet = pendingQueue.remove();
}
try {
if (packet.requestHeader.getXid() != replyHdr.getXid()) {
packet.replyHeader.setErr(
KeeperException.Code.CONNECTIONLOSS.intValue());
throw new IOException("Xid out of order. Got Xid "
+ replyHdr.getXid() + " with err " +
+ replyHdr.getErr() +
" expected Xid "
+ packet.requestHeader.getXid()
+ " for a packet with details: "
+ packet );
}
packet.replyHeader.setXid(replyHdr.getXid());
packet.replyHeader.setErr(replyHdr.getErr());
packet.replyHeader.setZxid(replyHdr.getZxid());
if (replyHdr.getZxid() > 0) {
lastZxid = replyHdr.getZxid();
}
if (packet.response != null && replyHdr.getErr() == 0) {
packet.response.deserialize(bbia, "response");
}
if (LOG.isDebugEnabled()) {
LOG.debug("Reading reply sessionid:0x"
+ Long.toHexString(sessionId) + ", packet:: " + packet);
}
} finally {
finishPacket(packet);
}
}

2.7　StaticHostProvider

提供服务端列表，客户端连接时会根据一定算法从列表中随机选择特定服务端连接。

2.7.1　updateServerList

根据是否新增Server的数量来决定平衡状态。

1. 首先看是否配置一定要重新连Server，如果是，则不管其他条件如何都挑选下一个可供选择的Server并连接。

2. 当前没连任何Server，则选择第一个或者最后一个Server连接。

3. 开始balance算法：

重新计算本次新增的Server和之前已存在的Server。

3　投票选举

ZooKeeper服务端集群在正常工作时需要一个Leader和无数个Follower，他们都对外提供统一的服务接口，客户端连接任意一台服务端发送命令其结果都是相同的，并且结果都是严格按照客户端提交读写命令的顺序来执行。

其中Leader就负责维持整个集群数据的一致性和写操作的顺序执行。

当然集群中总是可能发生故障，当某个Follower发生故障时，该Follower会重新启动选举流程并重新连接到Leader，成功连接后会同步Leader的数据，这样就算之前有写操作失败了，通过同步过程最后会恢复完整的数据库。Leader/Follower的数据同步过程会在后面详细论述。

集群通过投票选举过程确认同一时刻集群最多只会存在一个Leader。当有新节点加入集群时，新节点会启动FastLeaderElection过程来重新选举，寻找新的Leader。

3.1　QuorumPeer

每台服务端主机都会启动一个QuorumPeer进程，该进程通常是从QuorumPeerMain类中启动QuorumPeer的，在Linux下用ps命令可以看到该进程。

QuorumPeer共有4种工作模式：

- LOOKING：选举模式，启动FastLeaderElection

- LEADING：领导者模式，启动Leader

- FOLLOWING：跟随者模式，启动Follower

- OBSERVING：旁观者模式，启动Observer

QuorumPeer刚启动时首先设置为LOOKING工作模式，它会启动投票选举过程，调用FastLeadElection类，启动Messenger接收选举信息。

启动流程如下：

QuorumPeer知道集群各个机器的IP地址，当通过FastLeaderElection完成选举过程后会确定自己是Leader还是Follower，如果是Leader则调用lead方法进入领导者工作流程；如果是Follower则调用followLeader方法进入追随者工作流程。

QuorumPeer维持在4种状态之间不停切换。在LOOKING状态时有一个细节要关注一下，如果shuttingDownLE为true，则表示要关闭FastLeaderElection线程重新启动，这时候Messenger也会关闭不再接收集群的选举信息。

如果只是简单的调用lookForLeader方法，则只是重新一轮寻找Leader的过程并不会关闭Messenger，这时候Leader一般大部分情况下还是原来的Leader。

主循环代码如下：

while (running) {
switch (getPeerState()) {
case LOOKING:
LOG.info("LOOKING");
try {
reconfigFlagClear();
if (shuttingDownLE) {
shuttingDownLE = false;
startLeaderElection();
}
setCurrentVote(makeLEStrategy().lookForLeader());
} catch (Exception e) {
LOG.warn("Unexpected exception", e);
setPeerState(ServerState.LOOKING);
}
break;
case OBSERVING:
try {
LOG.info("OBSERVING");
setObserver(makeObserver(logFactory));
observer.observeLeader();
} catch (Exception e) {
LOG.warn("Unexpected exception",e );
} finally {
observer.shutdown();
setObserver(null);
updateServerState();
}
break;
case FOLLOWING:
try {
LOG.info("FOLLOWING");
setFollower(makeFollower(logFactory));
follower.followLeader();
} catch (Exception e) {
LOG.warn("Unexpected exception",e);
} finally {
follower.shutdown();
setFollower(null);
updateServerState();
}
break;
case LEADING:
LOG.info("LEADING");
try {
setLeader(makeLeader(logFactory));
leader.lead();
setLeader(null);
} catch (Exception e) {
LOG.warn("Unexpected exception",e);
} finally {
if (leader != null) {
leader.shutdown("Forcing shutdown");
setLeader(null);
}
updateServerState();
}
break;
}
}

QuorumPeer周而复始在这些状态间切换，维持Zookeeper集群的高可用性和一致性。

主要方法

名称

说明

createElectionAlgorithm

启动选举类线程，这里是FastLeaderElection，同时启动QuorumCnxManager建立Socket监听选举类消息。

startLeaderElection

启动选举过程

run

主线程，管理QuorumPeer循环在FastLeaderElection、Leader、Follower、Observer几种状态之间切换。

3.2　QuorumCnxManager

QuorumCnxManager是专门用于选举信息交换的Socket框架，Messenger接收的报文就是从QuorumCnxManager的接收队列中获取的。

QuorumCnxManager通常采用Netty框架负责底册Socket连接管理，这是一种非阻塞的Socket方式，提供Select方法在多个Socket之间轮询数据，先到先得的方式处理来自不同QuorumPeer的选举交换消息。

发送线程定义为SendWorker，接收线程定义为RecvWorker。对于每一个QuorumPeer连接都会创建一个SendWorkder和一个RecvWorker。

发送队列是一个HashMap，定义如下：

ConcurrentHashMap<Long, ArrayBlockingQueue<ByteBuffer>> queueSendMap;

每个机器在queueSendMap中占一条记录，用机器的sid作为key。

接收队列定义为：

ArrayBlockingQueue<Message> recvQueue;

接收队列统一用一个Queue表示，不同机器的消息全都放在这个队列中。

Messenger发送消息时调用QuorumCnxManager的toSend方法，将发送消息添加到queueSendMap中，这里我们看到只有发送给远方的消息才会真正放入queueSendMap，发给自己的消息直接放到接收队列recvQueue中。

public void toSend(Long sid, ByteBuffer b) {
if (self.getId() == sid) {
b.position(0);
addToRecvQueue(new Message(b.duplicate(), sid));
} else {
ArrayBlockingQueue<ByteBuffer> bq = new ArrayBlockingQueue<ByteBuffer>(SEND_CAPACITY);
ArrayBlockingQueue<ByteBuffer> oldq = queueSendMap.putIfAbsent(sid, bq);
if (oldq != null) {
addToSendQueue(oldq, b);
} else {
addToSendQueue(bq, b);
}
connectOne(sid);
}
}

3.2.1　发送线程

发送线程SendWorker中的run方法负责从发送队列queueSendMap中取消息并发送出去，完整的代码这里不贴出了，看看run中主要的发送逻辑：

while (running && !shutdown && sock != null) {
ByteBuffer b = null;
try {
ArrayBlockingQueue<ByteBuffer> bq = queueSendMap.get(sid);
if (bq != null) {
b = pollSendQueue(bq, 1000, TimeUnit.MILLISECONDS);
} else {
LOG.error("No queue of incoming messages for " +
"server " + sid);
break;
}
if(b != null){
lastMessageSent.put(sid, b);
send(b);
}
} catch (InterruptedException e) {
LOG.warn("Interrupted while waiting for message on queue", e);
}
}

取出本次发送的消息放入lastMessageSend队列中，然后调用send方法发送，send方法源码：

synchronized void send(ByteBuffer b) throws IOException {
byte[] msgBytes = new byte[b.capacity()];
try {
b.position(0);
b.get(msgBytes);
} catch (BufferUnderflowException be) {
LOG.error("BufferUnderflowException ", be);
return;
}
dout.writeInt(b.capacity());
dout.write(b.array());
dout.flush();
}

dout定义：DataOutputStream dout;是Socket写接口。

3.2.2　接收线程

接收线程RecvWorker。

接收线程的run方法：

while (running && !shutdown && sock != null) {
int length = din.readInt();
byte[] msgArray = new byte[length];
din.readFully(msgArray, 0, length);
ByteBuffer message = ByteBuffer.wrap(msgArray);
addToRecvQueue(new Message(message.duplicate(), sid));
}

接收线程从底层Socket收到报文后放到recvQueue队列中，等待Messenger调用pollRecvQueue方法获取消息。

public Message pollRecvQueue(long timeout, TimeUnit unit)
throws InterruptedException {
return recvQueue.poll(timeout, unit);
}

pollRecvQueue方法调用ArrayBlockingQueue的poll方法取队列的数据，poll方法可以指定timeout时间，超时没有数据则会抛出InterruptedException异常。

3.3　Messenger

Messenger定义在FastLeaderElection类中，专用于QuorumPeer之间选举消息的传递，交换toSend/Notification消息体。

它由FastLeaderElection负责启动，负责选举过程中QuorumPeer间传递选举消息。Messenger包括WorkerReceiver和WorkerSender两个子线程，一个负责接收QuorumPeer消息，另一个负责发送QuorumPeer消息。

WorkerSender使用ToSend结构发送消息，而WorkerReceiver使用Notification结构接收消息。toSend和Notification消息体的属性基本是一致性的。

Messenger底层采用QuorumCnxManager作为Socket连接池，每个QuorumPeer都会创建一个QuorumCnxManager，用于与集群其他所有的QuorumPeer的信息交互。

3.3.1　Notification类

接收Vote信息的消息接收体，在选举过程中集群交换的包都是Notification类型。主要属性有以下这些：

名称

说明

leader

Leader编号

sid

发送者地址

zxid

事务Id

state

状态

electionEpoch

已选举次数，越高权威越大

3.3.2　ToSend类

ToSend和Nofitication类似，发送Vote信息的消息定义体。

Notification和ToSend只在Messenger中处理，除此之外的地方看不到这两种数据结构。

3.3.3　WorkerReceiver

WorkerReceiver是接收线程，专门接收参与选举的服务器返回的通知，进行相应的处理。

从QuorumCnxManager获取网络接收到的Message包，然后组装成Notification消息，放入recvqueue队列，等待后续处理。

如果消息来自观察者或者追随者，则将当前的领导者的zxid和sid立即回复给对方；

如果消息来自另一个LOOKING状态的参与选举者，则将对方发来包里的zxid和sid等信息组装Nofitication消息，然后判断：

A. 如果接收者是Looking状态，则将Nofitication放入FastLeaderElection的recvqueue等待处理（这种情况下，如果对方也是Looking状态，则判断哪方的Epoch和Zxid最大，大的为新的Leader候选，小的抛弃）。

B. 如果接收者不是Looking状态而发送者是LOOKING，则将当前接收者认为的Leader发送给对方。

其他的包则忽略。

主要代码如下：

while (!stop) {
response = manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
if(self.getPeerState() == QuorumPeer.ServerState.LOOKING){
recvqueue.offer(n);
if((ackstate == QuorumPeer.ServerState.LOOKING)
&& (n.electionEpoch < logicalclock.get())){
Vote v = getVote();
QuorumVerifier qv = self.getQuorumVerifier();
ToSend notmsg = new ToSend(ToSend.mType.notification,
v.getId(),
v.getZxid(),
logicalclock.get(),
self.getPeerState(),
response.sid,
v.getPeerEpoch(),
qv.toString().getBytes());
sendqueue.offer(notmsg);
}
} else {
Vote current = self.getCurrentVote();
if(ackstate == QuorumPeer.ServerState.LOOKING){
QuorumVerifier qv = self.getQuorumVerifier();
ToSend notmsg = new ToSend( ToSend.mType.notification,
current.getId(),
current.getZxid(),
current.getElectionEpoch(),
self.getPeerState(),
response.sid,
current.getPeerEpoch(),
qv.toString().getBytes());
sendqueue.offer(notmsg);
}
}
}

3.3.4　WorkerSender

WorkerSender是发送线程，专门负责发送ToSend消息给参与选举的其他机器。

WorkerSender的流程比较简单，就是将ToSend消息放入发送队列，发送线程循环发送。

主要代码如下：

public void run() {
while (!stop) {
try {
ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
if(m == null) continue;
process(m);
} catch (InterruptedException e) {
break;
}
}
LOG.info("WorkerSender is down");
}
void process(ToSend m) {
ByteBuffer requestBuffer = buildMsg(m.state.ordinal(),
m.leader,
m.zxid,
m.electionEpoch,
m.peerEpoch,
m.configData);
manager.toSend(m.sid, requestBuffer);
}

manager就是指QuorumCnxManager。

3.4　快速选举

ZooKeeper的选举过程默认使用FastLeaderElection类，FastLeaderElection启动时启动Messenger收发选举信息。选举完成后选出1个Leader和若干Follower。

首先理解几个概念：

Epoch：投票周期，用于区分每一个round，每一次建立一个新的leader-follower关系，都会有一个唯一的epoch值去标识。就好像皇帝登基必须得有一个年号，与之前或之后的皇帝进行区分。刚启动是从文件读取保存的currentEpoch和acceptedEpoch值。默认是-1。

zxid：事务ID，表示Zookeeper当前写操作的序列号，确保写操作的按顺序执行。有一种场景是当Follower的最大zxid大于Leader的zxid时，Leader会发送TRUNC包给Follower截断多余的事务，保证和Leader数据一致。zxid会在ZxDataBase的loadDataBase方法中初始化。每次执行一次写事务zxid就会加1，默认为0。

选举规则：先判断双方的Epoch，留取大的Epoch；再看双方的zxid，留取大的zxid一方。最后的结果就是（Epoch<32|zxid）的一方会被推举为Leader。

集群间选举使用Messenger来负责选举信息的交互，Messenger底层使用QuorumCnxManager管理本机和集群其他机器之间的Socket数据传递，通信流程示意图如下所示。

集群中所有LOOKING状态的机器同时启动选举过程。选举结束则产生一个Leader和多个Follower。然后就是各Follower连接到Leader进行事务初始化，实现数据同步。

Observer不参与选举，但会接收Leader的信息同步要求。

QuorumPeer刚启动时会设置初始状态为LOOKING，然后启动FastLeaderElection选举过程，首先将自己作为Vote群发给其他QuorumPeer，群发的数据结构为Notification。同时启动Messenage的收发线程，调用lookForLeader方法发起一轮选举Leader的过程。

在Leader/Follower状态，如果接收到远方LOOKING节点寻找Leader而发送的Notification包，则Messenger会自动回复Leader信息给对方；在LOOKING状态，则Messenger会回复当前Vote节点（临时Leader）信息给对方。

默认选举规则规定当超过半数的QuorumPeer都认同同一个Leader时，选举过程结束，各QuorumPeer将自己设置为Leader、Follower或Observer。当然有兴趣的读者也可以尝试创建自己的选举规则类，只要实现QuorumVerifier即可。

进行Fast leader election的先决条件：

1. 每个QuorumPeer都知道其他QuorumPeer的ip地址，并知道QuorumPeer的总数。

2. 每个QuorumPeer一开始都是发起一个vote，选取自己为leader。向其他所有的QuorumPeer发送vote的notification，并等待回复。

3. 根据QuorumPeer的状态处理vote notification消息。

选举过程中产生的临时Leader成为Vote，当选举结束后最后一轮产生的Vote即成为新的Leader。

主要方法

名称

说明

checkLeader

检查是不是已经有合法的leader，没有则返回true。

updateProposal

更新潜在的leader信息

getVote

当前潜在的leader，组装成Vote对象

getInitId

初始化Id

lookForLeader

启动新一轮寻找Leader过程，只要状态改成LOOKING，该方法就会触发。

learningState

如果不是Leader，决定自己到底是Follower还是Observer

totalOrderPredicate

判断新的Vote是否大于自己的Vote，判断对方发过来的zxid临时选举结果是否更有效果。

termPredicate

是否可以结束选举过程，true说明已选出Leader

3.4.1　lookForLeader

QuorumPeer每次发起选举都调用lookForLeader方法实现，首先将自己设置为LOOKING状态。该方法是FastLeaderElection类的主方法，具体的流程如下：

（一）首先更新选举周期logicalclock，并把自己作为leader作为投票发给所有其他的server。

（二）然后进入本轮投票的循环，直到自己不再是LOOKING状态。

1. 从recvqueue获取一个网络包（recvqueue的数据来自Messenger），如果没有收到包则检查是否要重连和重发自己的投票。

2. 收到投票后判断对方投票的状态。

i.　LOOKING：

a) 如果对方投票的周期（Epoch）大于自己的周期（Epoch），那就清空自己的已经收到的投票集合recvset，并将自己作为候选和对方投票的leader做比较，选出大的作为新的投票，然后再发送给所有人。

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

c) 如果对方的投票周期小于自己，则忽略对方的投票。

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

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

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

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

iii.　FOLLOWING或LEADING：

a) 如果对方和自己再一个时钟周期，说明对方已经完成选举，如果对方说它是leader，那我们就把它作为leader，否则就要比较下对方选举的leader在自己这里是否占有多数，并且选举的leader确认了愿意当leader，如果都通过了，就把这个候选作为自己的leader

b) 如果对方和自己不再一个时钟周期，说明自己挂掉后又恢复起来，这个时候把别人的投票收集到一个单独的集合outofelection（从名字可以看出这个集合不是用在选举判断），如果对方的投票在outofelection中占有大多数，并且leader也确认了自己愿意做leader，这个时候更新自己的选举周期logicalclock，并修改自己的状态为FOLLOWING或LEADING

lookForLeader代码较长，我们先看看它的主体结构。

public Vote lookForLeader() throws InterruptedException {
try {
HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();
HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();
int notTimeout = finalizeWait;
synchronized(this){
logicalclock.incrementAndGet();
updateProposal(getInitId(), getInitLastLoggedZxid(),
getPeerEpoch());
}
sendNotifications();
while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){
Notification n = recvqueue.poll(notTimeout,
TimeUnit.MILLISECONDS);
if (self.getCurrentAndNextConfigVoters().contains(n.sid)) {
switch (n.state) {
case LOOKING:
{A代码}
break;
case OBSERVING:
LOG.debug("Notification from observer: " + n.sid);
break;
case FOLLOWING:
case LEADING:
{B代码}
break;
default:
LOG.warn("Notification state unrecoginized: " + n.state
+ " (n.state), " + n.sid + " (n.sid)");
break;
}
}
}
return null;
}
}

具体分析如下：

首先通过sendNotification方法告诉集群我在寻找Leader，集群中其他机器会在Messenger进程中接收到Notification，然后都会回复谁应该是当前Leader。

FastLeaderElection收集到足够的Notification消息，来判断到底谁才是合法的Leader。为此它对每条Notification消息进行下列判断：

A.　回复Notification的发送方也是LOOKING状态

如果回复Notification的发送方也是LOOKING状态，说明它还不知道最终的Leader是谁，这时候FastLeaderElection和发送方比较，看看谁的潜在Leader的Epoch和zxid最大，最大的设置为当前的候选Leader，然后将这个候选Leader广播出去，让发送方也能更改自己的后续Leader。同时判断自己的收集的Notification能否达到超过半数的条件从而决定最终的Leader，如果能决定则设置最终Leader，退出选举过程。

case LOOKING:
if (getInitLastLoggedZxid() == -1) {
LOG.debug("Ignoring notification as our zxid is -1");
break;
}
if (n.zxid == -1) {
LOG.debug("Ignoring notification from member with -1 zxid" + n.sid);
break;
}
// If notification > current, replace and send messages out
if (n.electionEpoch > logicalclock.get()) {
logicalclock.set(n.electionEpoch);
recvset.clear();
if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
updateProposal(n.leader, n.zxid, n.peerEpoch);
} else {
updateProposal(getInitId(),getInitLastLoggedZxid(),getPeerEpoch());
}
sendNotifications();
} else if (n.electionEpoch < logicalclock.get()) {
break;
} else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
proposedLeader, proposedZxid, proposedEpoch)) {
updateProposal(n.leader, n.zxid, n.peerEpoch); sendNotifications();
}
recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch,
n.peerEpoch));
if (termPredicate(recvset, new Vote(proposedLeader, proposedZxid,
logicalclock.get(), proposedEpoch))) {
while((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) !=
null){
if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
proposedLeader, proposedZxid, proposedEpoch)){
recvqueue.put(n);
break;
}
}
if (n == null) {
self.setPeerState((proposedLeader == self.getId()) ?
ServerState.LEADING: learningState());
Vote endVote = new Vote(proposedLeader, proposedZxid,
proposedEpoch);
leaveInstance(endVote);
return endVote;
}
}
break;

totalOrderPredicate方法用于判断对方发送过来的Vote是不是更新的Leader候选者,如果是的话则更新本地proposedLeader。主要代码如下：

protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
return ((newEpoch > curEpoch) ||
((newEpoch == curEpoch) &&  ((newZxid > curZxid) ||
((newZxid == curZxid) && (newId > curId)))));
}

termPredicate用于判断是否满足选出Leader条件，QuorumVerifier接口的实现，如果满足则设置Leader，并退出FastLeaderElection过程。主要代码如下：

private boolean termPredicate(HashMap<Long, Vote> votes, Vote vote) {
SyncedLearnerTracker voteSet = new SyncedLearnerTracker();
voteSet.addQuorumVerifier(self.getQuorumVerifier());
if (self.getLastSeenQuorumVerifier() != null
&& self.getLastSeenQuorumVerifier().getVersion() > self
.getQuorumVerifier().getVersion()) {
voteSet.addQuorumVerifier(self.getLastSeenQuorumVerifier());
}
for (Map.Entry<Long, Vote> entry : votes.entrySet()) {
if (vote.equals(entry.getValue())) {
voteSet.addAck(entry.getKey());
}
}
return voteSet.hasAllQuorums();
}

B.　回复Notification的发送方也是Leader或Follower

如果回复Notification的发送方是Leader或者Follower，则流程比较简单，将Notification消息保存到recvset中，并调用termPredicate方法判断是否能确定Leader并结束选举过程。

代码片段：

case FOLLOWING:
case LEADING:
if(n.electionEpoch == logicalclock.get()){
recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch,
n.peerEpoch));
if(termPredicate(recvset, new Vote(n.leader, n.zxid, n.electionEpoch,
n.peerEpoch, n.state))  && checkLeader(outofelection, n.leader,
n.electionEpoch)) {
self.setPeerState((n.leader == self.getId()) ?
ServerState.LEADING: learningState());
Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
leaveInstance(endVote);
return endVote;
}
}

4　事务处理

QuorumPeer在LEADING状态时启动Leader，Following状态时启动Follower，OBSERVING状态时启动Observer。

Leader和Follower工作过程中需要通过一定的机制共同维护一份集群事务，不管谁响应客户端请求，都需要将写事务扩散到集群中每台机器，保证每台服务器拥有相同的事务序列。

Leader和Follower通过QuorumPacket消息维护事务的一致性。Follower启动时首先要连接到Leader，同步写事务历史记录，然后才启动ZooKeeperServer提供服务给客户端。

Leader接收到Follower的连接请求时创建LearnerHandler，为每个Follower开启一个处理线程，交换QuorumPacket包。

4.1　QuorumPacket

注册过程中使用的数据包是QuorumPacket。在注册过程中的QuorumPacket包的类型主要有以下几种：

名称

说明

FOLLOWERINFO/OBSERVERINFO

Follower或Observer信息

LEADERINFO/NEWLEADER

Leader信息，Leader发给Follower

ACKEPOCH

Epoch确认，Follower发给Leader

TRUNC/SNAP/DIFF

事务同步，Leader发给Follower

PROPOSAL

写事务请求，Leader发给Follower，收到COMMIT后执行

ACK

确认包，Follower发给Leader

COMMIT

提交事务，Leader发给Follower包

REQUEST

写事务请求，Follower发给Leader

4.2　Learner

Learner是Follower和Observer的父类，定义了共同方法。

这里主要介绍request方法，当FollowerZooKeeperServer或ObserverZooKeeperServer接收到客户端的请求时，会调用request方法来将请求发送给Leader，通知Leader进行集群事务广播和确认，发给Leader的QuorumPacket是REQUEST类型。

4.2.1　request

当Follower或者Observer接收到客户端的写操作请求时，就会调用父类的request方法，此时生成一个REQUEST类型的QuorumPacket包发送给Leader，告诉Leader要执行什么命令，然后Leader就会通过自己的调用链来处理这个写操作请求（可参考LeaderZooKeeperServer的数据请求过程）。

在LeaderZooKeeperServer和FollowerZooKeeperServer中还会更详细的论述写操作流程。

强调一下当客户端连接的服务端是Leader或者Follower时执行流程是不一样的。

如果Leader接收到客户端写请求，则Leader会触发调用链到ProposalRequestProcessor处理单元；

如果Follower接收到客户端请求，则Follower首先会调用request方法发送REQUEST包给Leader，Leader会通过与该Follower对应的LearnerHandler线程接收到该REQUEST包，然后出发LeaderZooKeeperServer的submitLearnerRequest方法处理该REQUEST包。submitLearnerRequest方法调用Leader调用链，首先调用PrepRequestProcessor的processRequest方法，然后沿着Leader的数据处理流程往下执行。

不管客户端连接的是Leader还是Follower最终都要通过Leader来主持写操作集群间通知同步的过程，保证写操作在集群间的所有机器上都被执行。

代码如下：

void request(Request request) throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream oa = new DataOutputStream(baos);
oa.writeLong(request.sessionId);
oa.writeInt(request.cxid);
oa.writeInt(request.type);
if (request.request != null) {
request.request.rewind();
int len = request.request.remaining();
byte b[] = new byte[len];
request.request.get(b);
request.request.rewind();
oa.write(b);
}
oa.close();
QuorumPacket qp = new QuorumPacket(Leader.REQUEST, -1, baos
.toByteArray(), request.authInfo);
writePacket(qp, true);
}

4.2.2　注册到Leader

Follower或Observer启动时需要注册到Leader，注册过程全部完成后才能启动ZooKeeperServer为客户端提供服务。

注册分为以下3个主要步骤：

a) 调用connectToLeader方法连接到Leader。

b) 调用registerWithLeader方法注册到Leader，交换各自的sid、zxid和Epoch等信息，Leader以此决定事务同步的方式。

c) 调用SyncWithLeader跟Leader进行事务数据同步，处理SNAP/DIFF/TRUNC包。

这3个方法都定义在父类Learner类中，Learner是Follower和Observer的父类。

下面我们以Follower作为例子说明注册到Leader的完整流程。

- connectToLeader

connectToLeader方法功能较简单，创建Socket连接到Leader。该方法定义在Follower的父类Learner中。

for (int tries = 0; tries < 5; tries++) {
try {
sockConnect(sock, addr, Math.min(self.tickTime * self.syncLimit, remainingInitLimitTime));
sock.setTcpNoDelay(nodelay);
break;
} catch{}
}

最多可以尝试5次连接。连接成功后Leader会创建一个LearnerHandler专门处理与该Follower之间的QuorunPacket消息的传递。

- registerWithLeader

Follower连接Leader成功之后，马上调用registerWithLeader方法。

registerWithLeader方法首先发送FOLLOWERINFO包给Leader，告诉Leader自己的身份属性（Follower的zxid，sid）。

long lastLoggedZxid = self.getLastLoggedZxid();
QuorumPacket qp = new QuorumPacket();
qp.setType(pktType);
qp.setZxid(ZxidUtils.makeZxid(self.getAcceptedEpoch(), 0));
LearnerInfo li = new LearnerInfo(self.getId(), 0x10000,
self.getQuorumVerifier().getVersion());
ByteArrayOutputStream bsid = new ByteArrayOutputStream();
BinaryOutputArchive boa = BinaryOutputArchive.getArchive(bsid);
boa.writeRecord(li, "LearnerInfo");
qp.setData(bsid.toByteArray());
writePacket(qp, true);
readPacket(qp);

然后等待Leader回复的LEADINFO包，Follower侧的代码如下：

final long newEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());
if (qp.getType() == Leader.LEADERINFO) {
leaderProtocolVersion = ByteBuffer.wrap(qp.getData()).getInt();
byte epochBytes[] = new byte[4];
final ByteBuffer wrappedEpochBytes = ByteBuffer.wrap(epochBytes);
if (newEpoch > self.getAcceptedEpoch()) {
wrappedEpochBytes.putInt((int)self.getCurrentEpoch());
self.setAcceptedEpoch(newEpoch);
} else if (newEpoch == self.getAcceptedEpoch()) {
wrappedEpochBytes.putInt(-1);
} else {
throw new IOException("Leaders epoch, " + newEpoch + " is less than accepted epoch, " + self.getAcceptedEpoch());
}
QuorumPacket ackNewEpoch = new QuorumPacket(Leader.ACKEPOCH, lastLoggedZxid, epochBytes, null);
writePacket(ackNewEpoch, true);
return ZxidUtils.makeZxid(newEpoch, 0);
}

这里readPacket读取Leader回复的QuorumPacket（LEADERINFO包），获取Leader的Epoch和zxid值，并更新Follower的Epoch和zxid值，以Leader信息为准。

最后给Leader发ACKEPOCH包，告诉Leader这次Follower已经与Leader的zxid同步了。

这里acceptedEpoch就是Leader的Epoch。

整个resigerWithLeader流程如下图所示：

接下来Follower就要进入syncWithLeader方法了。

- SyncWithLeader

SyncWithLeader方法同步Leader的事务到Follower，该方法较长，这里分段介绍其整个过程。

首先读取同步数据包，主要代码如下：

QuorumPacket qp = new QuorumPacket();
readPacket(qp);
if (qp.getType() == Leader.SNAP){
zk.getZKDatabase().deserializeSnapshot(leaderIs);
}else if (qp.getType() == Leader. TRUNC) {
boolean truncated=zk.getZKDatabase().truncateLog(qp.getZxid());
if (!truncated) {
// not able to truncate the log
LOG.error("Not able to truncate the log " + Long.toHexString(qp.getZxid()));
System.exit(13);
}
}else if (qp.getType() == Leader.DIFF) {
LOG.info("Getting a diff from the leader 0x{}", ong.toHexString(qp.getZxid()));
snapshotNeeded = false;
}

同步方式分成3种：

- SNAP：快照模式，这种模式下Leader将整个完整数据库传给Follower。

- TRUNC：截断模式，这种模式表明Follower的数据比Leader还多，为了维持一致性需要将Follower多余的数据删除。

- DIFF：差异模式，说明Follower比Leader的事务少，需要给Follower补足，这时候Leader会将需要补充的事务生成PROPOSAL包和COMMIT包发给Follower执行。

当前面都执行完成后，进入下一步处理Leader的后续QuorumPacket消息。

下一步主要代码如下：

while (self.isRunning()) {
readPacket(qp);
switch(qp.getType()) {
case Leader.PROPOSAL:
packetsNotCommitted.add(pif);
break;
case Leader.COMMIT:
case Leader.COMMITANDACTIVATE:
pif = packetsNotCommitted.peekFirst();
if (!writeToTxnLog) {
zk.processTxn(pif.hdr, pif.rec);
packetsNotCommitted.remove();
} else {
packetsCommitted.add(qp.getZxid());
}
break;
case Leader.INFORM:
case Leader.INFORMANDACTIVATE:
if (!writeToTxnLog) {
// Apply to db directly if we haven't taken the snapshot
zk.processTxn(packet.hdr, packet.rec);
} else {
packetsNotCommitted.add(packet);
packetsCommitted.add(qp.getZxid());
}
break;
case Leader.UPTODATE:
if (isPreZAB1_0) {
zk.takeSnapshot();
self.setCurrentEpoch(newEpoch);
}
self.setZooKeeperServer(zk);
self.adminServer.setZooKeeperServer(zk);
break outerLoop;
case Leader.NEWLEADER:
if (snapshotNeeded) {
zk.takeSnapshot();
}
self.setCurrentEpoch(newEpoch);
writeToTxnLog = true;
isPreZAB1_0 = false;
writePacket(new QuorumPacket(Leader.ACK, newLeaderZxid, null, null), true);
break;
}
}

下一步Leader会先发送当前阻塞事务包给Follower，这里一般是PROPOSAL、COMMIT、INFORM这几类，这些是正常的写事务会涉及到的QuorumPacket，就按照写事务的正常逻辑执行就可以了。

发完之后Leader会发送NEWLEADER包，Follower收到NEWLEADER包后回复ACK给Leader。

最后Leader发UPTODATE包表示同步完成，Follower这时启动服务端并跳出本次循环，准备结束整个注册过程。

整个syncWithLeader流程如下图所示：

4.3　Follower

Follower是Learner的子类，Follower的启动方法就是followLeader。

4.3.1　followLeader

followLeader的主要代码片段如下：

connectToLeader(addr);
long newEpochZxid = registerWithLeader(Leader.FOLLOWERINFO);
syncWithLeader(newEpochZxid);
QuorumPacket qp = new QuorumPacket();
while (this.isRunning()) {
readPacket(qp);
processPacket(qp);
}

启动时首先与Leader同步数据，然后启动FollowerZooKeeperServer，在FollowerZooKeeperServer运行的同时，额外启动while循环等待Peer的QuorumPacket包，调用processPacket方法处理这些包。

4.3.2　processPacket

处理QuorumPeer传送的QuorumPacket，最主要是处理两种QuorumPacket：PROPOSAL和COMMIT。当然还有PING、COMMITANDACTIVATE等包类型，为便于简化梳理代码设计思路，这里就不再详述了。

该方法在收到Leader发送过来的QuorumPacket时被调用，主要是响应PROPOSAL和COMMIT两种类型的消息。

PROPOSAL是Leader将要执行的写事务命令；COMMIT是提交命令。Follower只有在收到COMMIT消息时才真正执行PROPOSAL命令。

同一个写事务命令会在Leader和多个Follower上都执行一次，保证集群数据的一致性。

代码片段：

case Leader.PROPOSAL:
TxnHeader hdr = new TxnHeader();
Record txn = SerializeUtils.deserializeTxn(qp.getData(), hdr);
if (hdr.getZxid() != lastQueued + 1) {
LOG.warn("Got zxid 0x"
+ Long.toHexString(hdr.getZxid())
+ " expected 0x"
+ Long.toHexString(lastQueued + 1));
}
lastQueued = hdr.getZxid();
fzk.logRequest(hdr, txn);
break;
case Leader.COMMIT:
fzk.commit(qp.getZxid());
break;

Follower收到PROPOSAL消息后调用FollowerZooKeeperServer的logRequest方法；收到COMMIT消息后调用FollowerZooKeeperServer的commit方法。方法的实现见下一章。

- PROPOSAL包

Leader发送过来的写操作指令，Leader执行写操作时需要告之集群中的Learner，让大家也执行写操作，保证集群数据的一致性。从代码中可以看到PROPOSAL是严格按照顺序执行的，这也是ZOOKEEPER的核心设计思想之一。Follower最终通过FollowerZooKeeperServer的logRequest方法先持久化Proposal，然后等待Leader发送COMMIT包之后才开始调用FollowerZooKeeperServer的commit方法提交该Proposal，最后调用到FinalRequestProcessor执行写操作。

- COMMIT包

当Leader认为一个Proposal已被大多数Follower持久化并等待执行后会发送COMMIT包，通知各Follower可以提交执行该Proposal了，通过这种机制保证写操作能被大半数集群机器执行。

4.4　Observer

Observer和Follower功能类似，主要的差别就是不参与选举。

Observer的入口方法是observerLeader。当QuorumPeer的状态是OBSERVING时会启动Observer并调用observerLeader方法。

4.4.1　observeLeader

同Follower的followLeader方法类似，首先注册到Leader，事务同步后进入QuorumPacket包循环处理过程，调用processPacket方法处理QuorumPacket。

void observeLeader() throws Exception {
zk.registerJMX(new ObserverBean(this, zk), self.jmxLocalPeerBean);
try {
InetSocketAddress addr = findLeader();
LOG.info("Observing " + addr);
try {
connectToLeader(addr);
long newLeaderZxid = registerWithLeader(Leader.OBSERVERINFO);
if (self.isReconfigStateChange())
throw new Exception("learned about role change");
syncWithLeader(newLeaderZxid);
QuorumPacket qp = new QuorumPacket();
while (this.isRunning()) {
readPacket(qp);
processPacket(qp);
}
} catch (Exception e) {
LOG.warn("Exception when observing the leader", e);
try {
sock.close();
} catch (IOException e1) {
e1.printStackTrace();
}
// clear pending revalidations
pendingRevalidations.clear();
}
} finally {
zk.unregisterJMX(this);
}
}

4.4.2　processPacket

比Follower要简单许多，最主要是处理INFORM包来执行Leader的写请求命令。

switch (qp.getType()) {
case Leader.INFORM:
TxnHeader hdr = new TxnHeader();
Record txn = SerializeUtils.deserializeTxn(qp.getData(), hdr);
Request request = new Request (hdr.getClientId(),  hdr.getCxid(), hdr.getType(), hdr, txn, 0);
ObserverZooKeeperServer obs = (ObserverZooKeeperServer)zk;
obs.commitRequest(request);
break;
}

这里处理的是INFORM消息，Leader群发写事务时，给Follower发的是PROPOSAL并要等待Follower确认；而给Observer发的则是INFORM消息并且不需要Obverver回复ACK消息来确认。

4.5　Leader

领导者就是Leader，是整个集群的写事务流程负责人。

一轮选举结束时产生新的Leader，并且Epoch加1。同时新的Leader先将自己的zxid设置为Epoch左移32位（Epoch<32），将是集群中最大的zxid。

Leader监听Socket等待Follower的连接请求，每次新的Follower连接的时候都会启动一个LearnerHandler线程专门处理与该Follower的交互。LearnerHandler循环接收Follower的消息包，并交给Leader进行处理。

- 启动

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

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

3. 调用Leader::lead函数，执行Leader的真正的逻辑

a) 调用ZooKeeperServer::loadData，从磁盘中恢复数据和session列表

b) 启用新的epoch，zookeeper中的zxid是64位，用于唯一标示一个操作，zxid的高32位是epoch，每次Leader切换加1，低32位为序列号，每次操作加1

c) 启动绑定在QuorumAddress上的Server，为每个Follower的连接建立一个LearnerHandler，用于和Follower做交互，这里的逻辑另外单独论述

d) 向所有的Follower发送一个NEWLEADER包，宣告自己额Leader身份，并在initLimit时间内等待大多数的Follower完成和Leader的同步，并发送ACK包，表示Follower已经和Leader完成同步并可以对外提供服务

e) 这时Leader和Client之间的交互在cnxnFactory的Server中，Leader和Follower之间的交互在LearnerHandler所属的线程中

f) 然后调用Leader::lead函数的QuromPeer线程在每个tickTime中都会发送ping消息给其他的follower，follower在接收到ping消息后会回复一个ping消息，并附带上follower的session tracker里的所有session信息，leader收到follower的ping消息后，根据传回的session信息更新自己的session信息 。

- Leader的主要方法

名称

说明

lead

主函数，执行领导者任务：
（1）导入Zk数据库数据，启动监听线程LearnerCnxAcceptor
（2）为每个Socket创建LearnerHandler，处理Follower注册过程
（3）等待超过半数的Follower连接成功
（4）LearnerHandler运行中，定时给Follower发送PING心跳

shutdown

按顺序关闭监听Socket，ZKServer，跟随者们Learners。

sendPacket

发送投票包给所有Follower

sendObserverPacket

发送投票包给所有Observer

commit

发送确认包给所有Follower，调用sendPacket

inform

发送inform包给所有Observer

proposal

发送提议包给参与者们

processSync

处理同步请求，发送同步包给客户端，调用sendSync方法

sendSync

发送同步包

4.5.1　响应Follower注册

Leader在接收到Follower的连接请求后（Follower调用connectToLeader方法），等待Follower发送第一个包，如果是Follower则第一个包是FOLLOWERINFO类型，这里为了简化我们不讨论Observer的情况。

收到到FOLLOWERINFO：

QuorumPacket qp = new QuorumPacket();
ia.readRecord(qp, "packet");
if (qp.getType() == Leader.OBSERVERINFO) {
learnerType = LearnerType.OBSERVER;
}
long lastAcceptedEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());
long peerLastZxid;
StateSummary ss = null;
long zxid = qp.getZxid();
long newEpoch = leader.getEpochToPropose(this.getSid(), lastAcceptedEpoch);
long newLeaderZxid = ZxidUtils.makeZxid(newEpoch, 0);
QuorumPacket newEpochPacket = new QuorumPacket(Leader.LEADERINFO, newLeaderZxid, ver, null);
oa.writeRecord(newEpochPacket, "packet");
bufferedOutput.flush();
QuorumPacket ackEpochPacket = new QuorumPacket();
ia.readRecord(ackEpochPacket, "packet");
if (ackEpochPacket.getType() != Leader.ACKEPOCH) {
LOG.error(ackEpochPacket.toString() + " is not ACKEPOCH");
return;
}
ByteBuffer bbepoch = ByteBuffer.wrap(ackEpochPacket.getData());
ss = new StateSummary(bbepoch.getInt(), ackEpochPacket.getZxid());
leader.waitForEpochAck(this.getSid(), ss);
peerLastZxid = ss.getLastZxid();

1. lastAcceptedEpoch：是Follower的Epoch值。

2. Zxid：是Follower的zxid

3. newEpoch：Leader根据FOLLOWERINF的值计算出新的Epoch

4. newLeaderZxid：根据新的Epoch计算新的Leader的zxid

然后给Follower发送LEADERINFO包，将新的zxid告诉Follower，等待Follower的ACKEPOCH回复包。收到ACKEPOCH包后，两者说明已同步了Epoch。

Leader然后判断是否要发SNAP包给Follower，如果需要则直接发SNAP包。

如果Follower的事务比Leader少一些（在minCommittedLog 和maxCommittedLog之间），则不需发SNAP包，而是发DIFF包，同时将需补充的事务通过PROPOSAL和COMMIT发给Follower执行。相关逻辑在syncFollower，queueCommittedProposals，startSendingPackets等方法中实现。这部分主要代码如下所示：

boolean needSnap = syncFollower(peerLastZxid, leader.zk.getZKDatabase(), leader);
QuorumPacket newLeaderQP = new QuorumPacket(Leader.NEWLEADER, newLeaderZxid,leader.self.getLastSeenQuorumVerifier().toString().getBytes(), null);
queuedPackets.add(newLeaderQP);
if (needSnap) {
try {
long zxidToSend = leader.zk.getZKDatabase().getDataTreeLastProcessedZxid();
oa.writeRecord(new QuorumPacket(Leader.SNAP, zxidToSend, null, null), "packet");
bufferedOutput.flush();
leader.zk.getZKDatabase().serializeSnapshot(oa);
oa.writeString("BenWasHere", "signature");
bufferedOutput.flush();
} finally {
snapshot.close();
}
}
startSendingPackets();

startSendingPackets将需要同步的事务发送给Follower，事务同步完成后，Leader发送NEWLEADER包给Follower。

然后等Follower回复第一个ACK包。收到ACK之后调用Leader的waitForNewLeaderAck方法告诉Leader该Follower已经完成同步。

当Leader收到足够多的waitForNewLeaderAck方法调用后（通常超过半数），知道大部分Follower已经注册到本Leader上来了，这时候Leader才能确保正式发挥Leader的作用了。

这些全部完成之后Leader发送UPTODATE包给Follower。通知Follower数据全部更新完毕。

主要代码逻辑如下：

qp = new QuorumPacket();
ia.readRecord(qp, "packet");
if(qp.getType() != Leader.ACK){
LOG.error("Next packet was supposed to be an ACK,"
+ " but received packet: {}", packetToString(qp));
return;
};
leader.waitForNewLeaderAck(getSid(), qp.getZxid(), getLearnerType());
queuedPackets.add(new QuorumPacket(Leader.UPTODATE, -1, null, null));

4.5.2　事务处理

当Leader收到客户端的写事务请求时，会发送PROPOSAL消息给Follower，通知Follower准备执行该写事务。

相反，如果是Follower接收到客户端的写事务请求，则Follower会发送REQUEST消息给Leader，让Leader来主持该写事务请求在集群间群发，最终目的都是保证写操作在集群间所有机器都能被同步执行。

Leader和Follower之间数据交互流程示意图如下：

- LearnerHandler主循环

注册和事务数据同步完成后，LearnerHandler进入循环处理QuorumPacket过程。

LearnerHandler正常处理的QurumPacket包主要的有以下这么几类：

PING：心跳包，回复一个PING给对方即可

ACK：当Follower收到PROPOSAL并写到磁盘后会发送ACK给Leader，详细流程见下章。

REQUEST：当Follower收到客户端写命令请求时，会发REQUEST给Leader，代码如下。

while (true) {
qp = new QuorumPacket();
ia.readRecord(qp, "packet");
ByteBuffer bb;
long sessionId;
int cxid;
int type;
switch (qp.getType()) {
case Leader.ACK:
if (this.learnerType == LearnerType.OBSERVER) {
if (LOG.isDebugEnabled()) {
LOG.debug("Received ACK from Observer  " + this.sid);
}
}
syncLimitCheck.updateAck(qp.getZxid());
leader.processAck(this.sid, qp.getZxid(),
sock.getLocalSocketAddress());
break;
case Leader.PING:
// Process the touches
ByteArrayInputStream bis = new ByteArrayInputStream(qp
.getData());
DataInputStream dis = new DataInputStream(bis);
while (dis.available() > 0) {
long sess = dis.readLong();
int to = dis.readInt();
leader.zk.touch(sess, to);
}
break;
case Leader.REQUEST:
bb = ByteBuffer.wrap(qp.getData());
sessionId = bb.getLong();
cxid = bb.getInt();
type = bb.getInt();
bb = bb.slice();
Request si;
if(type == OpCode.sync){
si = new LearnerSyncRequest(this, sessionId, cxid, type, bb, qp.getAuthinfo());
} else {
si = new Request(null, sessionId, cxid, type, bb, qp.getAuthinfo());
}
si.setOwner(this);
leader.zk.submitLearnerRequest(si);
break;
default:
LOG.warn("unexpected quorum packet, type: {}", packetToString(qp));
break;
}
}

- proposal

当LeaderZooKeeperServer收到客户端写事务请求时，会触发Leader的proposal方法执行，发送PROPOSAL消息给Follower。

Follower收到PROPOSAL消息后首先保存到磁盘，然后回复ACK消息给Leader。

Leader收到足够多（超过半数）的ACK消息后会发送COMMIT消息给所有的Follower，告诉它们可以执行真正的写事务了，同时Leader会在本地沿调用链最终到达FinalRequestProcessor去执行本次写操作。同样的，Follower收到COMMIT包后也会真正的调用到FinalRequestProcessor在Follower本机执行写事务。

Leader和Follower都执行了相同的写事务命令，保证了集群间的数据一致性。

proposal方法的主要代码如下：

ByteArrayOutputStream baos = new ByteArrayOutputStream();
BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
try {
request.getHdr().serialize(boa, "hdr");
if (request.getTxn() != null) {
request.getTxn().serialize(boa, "txn");
}
baos.close();
} catch (IOException e) {
LOG.warn("This really should be impossible", e);
}
QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid,
baos.toByteArray(), null);
Proposal p = new Proposal();
p.packet = pp;
p.request = request;
synchronized(this) {
p.addQuorumVerifier(self.getQuorumVerifier());
if (request.getHdr().getType() == OpCode.reconfig){
self.setLastSeenQuorumVerifier(request.qv, true);
}
lastProposed = p.packet.getZxid();
outstandingProposals.put(lastProposed, p);
sendPacket(pp);
}

- inform

和proposal类似，只不过是将写事务发给Observer，并且不需要Observer回复ACK。

public void inform(Proposal proposal) {
QuorumPacket qp = new QuorumPacket(Leader.INFORM,
proposal.request.zxid, proposal.packet.getData(), null);
sendObserverPacket(qp);
}

- processAck

Follower持久化PROPOSAL写事务请求到磁盘后会回复ACK消息给Leader，Leader通过LearnerHandler接收该包，并触发Leader的processAck方法。

Leader发送到Follower的PROPOSAL消息，需等待所有Follower回复Ack消息后判断是否满足COMMIT条件，如果满足COMMIT条件则发送COMMIT消息给所有Follower，Leader再往下执行。

synchronized public void processAck(long sid, long zxid, SocketAddress followerAddr) {
if (!allowedToCommit) return;
if ((zxid & 0xffffffffL) == 0) {
return;
}
if (outstandingProposals.size() == 0) {
if (LOG.isDebugEnabled()) {
LOG.debug("outstanding is 0");
}
return;
}
if (lastCommitted >= zxid) {
return;
}
Proposal p = outstandingProposals.get(zxid);
if (p == null) {
LOG.warn("Trying to commit future proposal: zxid 0x{} from {}",
Long.toHexString(zxid), followerAddr);
return;
}
p.addAck(sid);
boolean hasCommitted = tryToCommit(p, zxid, followerAddr);
if (hasCommitted && p.request!=null && p.request.getHdr().getType() == OpCode.reconfig){
long curZxid = zxid;
while (allowedToCommit && hasCommitted && p!=null){
curZxid++;
p = outstandingProposals.get(curZxid);
if (p !=null)
hasCommitted = tryToCommit(p, curZxid, null);
}
}
}

ProcessAck会统计多少Follower回复了Leader.ACK类型的QuorumPacket包，然后就走到了tryToCommit方法，tryToCommit判断是否满足COMMIT条件。

下面看看tryToCommit方法执行了哪些操作。

- tryToCommit

确保写操作是按顺序被确认的。

尝试对Proposal按顺序进行Commit。Commit过的事务才是真正有效的事务。

事务确认必须按顺序进行，outstandingProposals中记录了所有等待确认的事务，只要前一条事务还未确认，则之后的事务都禁止确认，以确保事务的按顺序进行。

确认过的事务放入toBeApplied队列中等待下一步处理（也就是FinalRequestProcessor）。

if (outstandingProposals.containsKey(zxid - 1))
return false;
if (!p.hasAllQuorums()) {
return false;
}
// commit proposals in order
if (zxid != lastCommitted+1) {
LOG.warn("Commiting zxid 0x" + Long.toHexString(zxid)
+ " from " + followerAddr + " not first!");
LOG.warn("First is "  + (lastCommitted+1));
}
outstandingProposals.remove(zxid);
if (p.request != null) {
toBeApplied.add(p);
}
if (p.request == null) {
LOG.warn("Going to commmit null: " + p);
} else if (p.request.getHdr().getType() == OpCode.reconfig) {
Long designatedLeader = getDesignatedLeader(p, zxid);
QuorumVerifier newQV = p.qvAcksetPairs.get(p.qvAcksetPairs.size()-1).getQuorumVerifier();
self.processReconfig(newQV, designatedLeader, zk.getZxid(), true);
if (designatedLeader != self.getId()) {
allowedToCommit = false;
}
commitAndActivate(zxid, designatedLeader);
informAndActivate(p, designatedLeader);
} else {
commit(zxid);
inform(p);
}
zk.commitProcessor.commit(p.request);
if(pendingSyncs.containsKey(zxid)){
for(LearnerSyncRequest r: pendingSyncs.remove(zxid)) {
sendSync(r);
}
}

- commit

通知Follower提交proposal执行。

通常只有当Leader从Follower收到超过半数（默认算法，当然你也可以实现自己的判断逻辑，比如改成超过2/3人数）的ACK才会发起commit，发送Leader.COMMIT包给Follower。

public void commit(long zxid) {
synchronized(this){
lastCommitted = zxid;
}
QuorumPacket qp = new QuorumPacket(Leader.COMMIT, zxid, null, null);
sendPacket(qp);
}

Follower接收到Leader.COMMIT包后会进入CommitProcessor处理链。

5　服务端

服务端专门用于响应ZooKeeper客户端的请求，类名称为ZooKeeperServer。

它有3个子类，名称为：LeaderZooKeeperServer，FollowerZooKeeperServer，ObserverZooKeeperServer。顾名思义它们分别在Leader，Follower，Observer时启动。

QuorumPeer运行在Leader模式时，启动LeaderZooKeeperServer为客户端提供服务。当LeaderZooKeeperServer收到客户端写事务请求时，先将请求交给Leader来发起整个集群的写事务流程，这时候Leader给所有Follower发送PROPOSAL，要求Follower将写请求持久化到本地磁盘以防丢失，Follower持久化后回复ACK信息给Leader，当Leader收集到超过半数的ACK回复时，给Follower发送COMMIT包提交事务执行，最终Leader和Follower都会执行该事务。当然最后是由LeaderZooKeeperServer返回结果给客户端的。

QuorumPeer运行在Follower模式时，启动FollowerZooKeeperServer为客户端提供服务。这时交给Follower来处理，Follower会发送REQUEST给Leader。Leader再给所有Follower发送PROPOSAL发起写事务流程，后面的流程和Leader模式下类似。

5.1　启动类

服务端的启动类是QuorumPeerMain。QuorumPeerMain中读取配置信息并启动QuorumPeer。配置信息由QuorumPeerConfig类负责读取。

QuorumPeerMain的主要启动代码如下：

public void runFromConfig(QuorumPeerConfig config)  throws IOException, AdminServerException{
LOG.info("Starting quorum peer");
try {
ServerCnxnFactory cnxnFactory = null;
ServerCnxnFactory secureCnxnFactory = null;
if (config.getClientPortAddress() != null) {
cnxnFactory = ServerCnxnFactory.createFactory();
cnxnFactory.configure(config.getClientPortAddress(),
config.getMaxClientCnxns(),false);
}
if (config.getSecureClientPortAddress() != null) {
secureCnxnFactory = ServerCnxnFactory.createFactory();              secureCnxnFactory.configure(config.getSecureClientPortAddress(),  config.getMaxClientCnxns(),true);
}
quorumPeer = getQuorumPeer();
quorumPeer.setTxnFactory(new FileTxnSnapLog(
config.getDataLogDir(),
config.getDataDir()));
quorumPeer.enableLocalSessions(config.areLocalSessionsEnabled());
quorumPeer.enableLocalSessionsUpgrading(
config.isLocalSessionsUpgradingEnabled());
quorumPeer.setElectionType(config.getElectionAlg());
quorumPeer.setMyid(config.getServerId());
quorumPeer.setTickTime(config.getTickTime());
quorumPeer.setMinSessionTimeout(config.getMinSessionTimeout());
quorumPeer.setMaxSessionTimeout(config.getMaxSessionTimeout());
quorumPeer.setInitLimit(config.getInitLimit());
quorumPeer.setSyncLimit(config.getSyncLimit());
quorumPeer.setConfigFileName(config.getConfigFilename());
quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
quorumPeer.setQuorumVerifier(config.getQuorumVerifier(), false);
if (config.getLastSeenQuorumVerifier()!=null) {
quorumPeer.setLastSeenQuorumVerifier(config.getLastSeenQuorumVerifier(), false);
}
quorumPeer.initConfigInZKDatabase();
quorumPeer.setCnxnFactory(cnxnFactory);
quorumPeer.setSecureCnxnFactory(secureCnxnFactory);
quorumPeer.setLearnerType(config.getPeerType());
quorumPeer.setSyncEnabled(config.getSyncEnabled());
quorumPeer.setQuorumListenOnAllIPs(config.getQuorumListenOnAllIPs());
quorumPeer.start();
quorumPeer.join();
} catch (InterruptedException e) {
// warn, but generally this is ok
LOG.warn("Quorum Peer interrupted", e);
}
}

shutdownLatch.await()等待ZooKeeperServer关闭或者发生异常，此时执行关闭代码。Await()使用了java的CountDownLatch机制来实现线程同步。在ERROR和SHUTDOWN状态时触发。

概括说，ZooKeeperServerMain启动ZooKeeperServer并等待关闭事件，触发时先调用自己的shutdown（）函数，然后调用ZooKeeperServer的shutdown（）事件，等待Socket连接线程结束，如果ZooKeeperServer可以shutdown则显式调用ZooKeeperServer的shutdown事件（感觉这里有点绕，一般SHUTDOWN事件也是ZooKeeperServer触发的，这时候ZooKeeperServer通常可以自己执行shutdown了，这里这么处理当然也没问题，更可靠吧）。

启动的ServerCnxn创建Socket并接收网络包，包传递给ZooKeeperServer处理，ZooKeeperServer先保存包到Queue然后用处理线程来处理包，处理包的时候执行RequestProcessor处理链。

5.2　ZooKeeperServer

ZooKeeperServer是服务端基类，它使用NettyServerCnxn作为底层Socket通信机制，NettyServerCnxn收到客户端请求之后会自动调用ZooKeeperServer的prorcessPacket方法来执行客户端命令，并发送响应包给客户端。

ZooKeeper服务端定义了3种子类：LeaderZooKeeperServer、FollowerZooKeeperServer和ObserverZooKeeperServer。它们分别用于Leader，Follower和Observer。

ZooKeeper客户端连接服务端时，可能连接的是Leader，也可能连接的是Follower或者Observer。连接到不同类型的服务端它的命令请求流程是不太一样的，特别是写事务请求。

如果连接的是Leader，则Leader会将写请求生成Proposal包广播到Follower，通知Follower有写事务要执行，这时Follower先将写事务请求保存到磁盘，然后给Leader发送ACK确认。Leader在超过半数确认的情况下才会真正开始执行写命令，写命令完成后Leader再给Follower发送COMMIT命令，通知Follower写操作已完成，这时Follower也会接着调用CommitProcessor并且最后调用FinalRequestProcessor执行同样的写命令，全部完成后Leader和Follower仍然保持相同的ZooKeeper事务，保持集群数据一致性。

读操作则不需要这么复杂，客户端不管连接的是Leader还是Follower，它发送的读命令都会直接执行并返回结果给客户端，因此读操作的并发性能要高很多。

ZooKeeper客户端和服务端之间数据交互流程示意图如下：

它们之间采用Netty框架作为底层Socket连接，NettyServerFactory是服务端连接管理池，为每个客户端创建一个NettyServer，NettyServer接收到客户端命令请求之后会调用ZooKeeperServer的processPacket方法处理该客户端请求。

5.2.1　客户端请求

客户端请求到达时会调用processPacket方法，processPacket接着调用submitRequest，submitRequest接着调用firstProcessor处理请求。

源码贴出：

public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException {
InputStream bais = new ByteBufferInputStream(incomingBuffer);
BinaryInputArchive bia = BinaryInputArchive.getArchive(bais);
RequestHeader h = new RequestHeader();
h.deserialize(bia, "header");
incomingBuffer = incomingBuffer.slice();
if (h.getType() == OpCode.auth) {
LOG.info("got auth packet " + cnxn.getRemoteSocketAddress());
AuthPacket authPacket = new AuthPacket();
ByteBufferInputStream.byteBuffer2Record(incomingBuffer, authPacket);
String scheme = authPacket.getScheme();
ServerAuthenticationProvider ap = ProviderRegistry.getServerProvider(scheme);
Code authReturn = KeeperException.Code.AUTHFAILED;
if (authReturn == KeeperException.Code.OK) {
if (LOG.isDebugEnabled()) {
LOG.debug("Authentication succeeded for scheme: " + scheme);
}
LOG.info("auth success " + cnxn.getRemoteSocketAddress());
ReplyHeader rh = new ReplyHeader(h.getXid(), 0,
KeeperException.Code.OK.intValue());
cnxn.sendResponse(rh, null, null);
} else {
ReplyHeader rh = new ReplyHeader(h.getXid(), 0,
KeeperException.Code.AUTHFAILED.intValue());
cnxn.sendResponse(rh, null, null);
cnxn.sendBuffer(ServerCnxnFactory.closeConn);
cnxn.disableRecv();
}
return;
} else {
if (h.getType() == OpCode.sasl) {
Record rsp = processSasl(incomingBuffer,cnxn);
ReplyHeader rh = new ReplyHeader(h.getXid(), 0, KeeperException.Code.OK.intValue());
cnxn.sendResponse(rh,rsp, "response");
} else {
Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(),
h.getType(), incomingBuffer, cnxn.getAuthInfo());
si.setOwner(ServerCnxn.me);
setLocalSessionFlag(si);
submitRequest(si);
}
}
cnxn.incrOutstandingRequests(h);
}

5.2.2　主要方法

名称

说明

takeSnapshot

将Zookeeper树结构写到磁盘，事务日志持久化支持。

processConnectRequest

处理连接请求，被NettyServerCnxn或NIOServerCnxn在收到网络连接请求时调用，他们都实现ServerCnxn接口。
ServerCnxn定义了一系列网络接口，这部分内容不再详细说明，本身和Zookeeper协议关系不大，是底层支持协议的范畴了。
该方法只处理创建Socket连接请求，具体的数据请求并不发生在这里。
为每个连接创建session

processPacket

处理Zookeeper数据服务请求，和processConnectRequest不同的是这次是已经建立起了连接，接收到客户端发来的请求信息。

setState

设置ZkServer运行状态，只有两种状态：RUNNING和SHUTDOWN。在SHUTDOWN时将关闭ZKServer，并通知ZKServerMain线程结束自己，调用清扫代码。

setupRequestProcessors

设置处理函数链条，按链条执行。

processPackage

ZkServer从ServerCnxn接收的包先判断是否是认证信息，认证失败则禁止该Socket接收后续数据。通过则调用submitRequest

submitRequest

判断ZookeeperServer的处理包流程是否准备就绪，没有则等待1秒再看是否抛出异常，通过则调用firstProcessor处理该包。
根据ZooKeeperServer子类的不同firstPrcessor具体指向也不同，Leader会指向LeaderRequestProcessor，Follower会指向FollowerRequestProcessor，具体流程后面会有专门论述。

processTxn

处理客户端写操作请求

5.2.3　processPacket

当ZooKeeperServer的底层NettyServerCnxn接收到客户端的Request时，该方法被NettyServerCnzx的主接收线程调用。ZooKeeperServer然后开始处理Request。

该方法的主要代码：

public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException {
// We have the request, now process and setup for next
InputStream bais = new ByteBufferInputStream(incomingBuffer);
BinaryInputArchive bia = BinaryInputArchive.getArchive(bais);
RequestHeader h = new RequestHeader();
h.deserialize(bia, "header");
Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(),
h.getType(), incomingBuffer, cnxn.getAuthInfo());
si.setOwner(ServerCnxn.me);
setLocalSessionFlag(si);
submitRequest(si);
}

从NettyServerCnxn收到Request，调用submitRequest方法。

public void submitRequest(Request si) {
if (firstProcessor == null) {
touch(si.cnxn);
boolean validpacket = Request.isValid(si.type);
if (validpacket) {
firstProcessor.processRequest(si);
if (si.cnxn != null) {
incInProcess();
}
} else {
LOG.warn("Received packet at server of unknown type " + si.type);
new UnimplementedRequestProcessor().processRequest(si);
}
}
}

调用firstProcessor开始处理Request，在Leader/Follower等不同的模式下其firstProcessor是不同的。

5.2.4　processTxn

处理客户端的写操作请求，通过ZKDatabase类实现具体逻辑。

该方法由FinalRequestProcessor调用，客户端读操作不在这里实现，而是放在FinalRequestProcessor中直接代码实现。当然读操作最终也是通过ZkDatabase的相关方法实现具体逻辑的。

处理写操作请求，写操作涉及到集群状态同步，和读操作有很大的区别，写操作提供了统一的入口方法processTxn来处理。

该方法的主要代码如下：

private ProcessTxnResult processTxn(Request request,TxnHeader hdr, Record txn) {
ProcessTxnResult rc;
int opCode = request != null ? request.type : hdr.getType();
long sessionId = request != null ? request.sessionId : hdr.getClientId();
if (hdr != null) {
rc = getZKDatabase().processTxn(hdr, txn);
} else {
rc = new ProcessTxnResult();
}
if (opCode == OpCode.createSession) {
if (hdr != null && txn instanceof CreateSessionTxn) {
CreateSessionTxn cst = (CreateSessionTxn) txn;
sessionTracker.addGlobalSession(sessionId, cst.getTimeOut());
} else if (request != null && request.isLocalSession()) {
request.request.rewind();
int timeout = request.request.getInt();
request.request.rewind();
sessionTracker.addSession(request.sessionId, timeout);
} else {
LOG.warn("*****>>>>> Got " + txn.getClass() + " " + txn.toString());
}
} else if (opCode == OpCode.closeSession) {
sessionTracker.removeSession(sessionId);
}
return rc;
}

5.3　调用链

服务端有两种消息要处理。一种是Request，表示客户端发来的读写操作请求；另一种是QuorumPacket，是Leader和Follower之间事务一致性相关的消息。

Request消息的处理分散在LeaderZooKeeperServer，FollowerZooKeeperServer，ObserverZooKeeperServer中；而QuorumPakcet消息的处理分散在Leader，Follower，Observer等类中。

本章的调用链是指Request消息的处理。

ZooKeeperServer收到客户端的操作请求后生成Request，然后是通过调用链的方式来处理Request。调用链定义了处理单元的链式执行。每个处理单元继承了Processor接口，并且可设置nextProcessor。

服务端调用链都是从firstProcessor开始往下顺序执行，每个调用模块执行完成之后沿着nextProcessor的方向依次执行下一个调用处理单元。

不同类型的ZooKeeperServer的调用链是不同的。主要有三种不同的调用链方式：

A. LeaderZooKeeperServer的调用链，它的firstProcessor是LeaderRequestProcessor。 B. FollowerZooKeeperServer的调用链，它的firstProcessor是FollowerRequestProcessor。

C. Observer ZooKeeperServer的调用链，它的firstProcessor是ObserverRequestProcessor。

ZooKeeperServer中定义的调用单元有以下这些类，它们都实现了Processor接口的processRequest方法。

调用单元名称

说明

LeaderRequesrProcessor

Leader的第一个processor

FollowerRequesrProcessor

Follower的第一个processor

ObserverRequesrProcessor

Observer的第一个processor

ProposalProcessor

命令提议，群发Proposal到Follower

CommitProcessor

事务提交，

SyncRequestProcessor

持久化Proposal到磁盘

AckRequestProcessor

Leader给自己ACK消息

SendAckRequestProcessor

Learner回复ACK给Leader

toBeAppliedProcessor

等待执行

PrepRequestProcessor

Leader使用，定义在LeaderRequestProcessor和ProposalRequestProcessor之间。
将写操作记录到outstandingChanges，在FinalRequestProcessor中确保按顺序执行Request，防止后续读操作在前一个写操作还未完成的情况下就被调用。

FinalRequestProcessor

最后的processor，调用ZkDatabase方法执行最终读写操作

下面我们来具体看看各个Processor的实现逻辑。

5.3.1　LeaderRequestProcessor

主要处理过程是查看是否存在upgradeSession的处理请求，这里我们不展开讨论upgradeSession的含义，感兴趣的读者可自行研究相关源码。

源码如下：

public void processRequest(Request request) throws RequestProcessorException {
Request upgradeRequest = null;
try {
upgradeRequest = lzks.checkUpgradeSession(request);
} catch (KeeperException ke) {
if (request.getHdr() != null) {
LOG.debug("Updating header");
request.getHdr().setType(OpCode.error);
request.setTxn(new ErrorTxn(ke.code().intValue()));
}
request.setException(ke);
LOG.info("Error creating upgrade request " + ke.getMessage());
} catch (IOException ie) {
LOG.error("Unexpected error in upgrade", ie);
}
if (upgradeRequest != null) {
nextProcessor.processRequest(upgradeRequest);
}
nextProcessor.processRequest(request);
}

5.3.2　FollowerRequestProcessor

Follower的调用链第一个处理单元，FollowerZooKeeper的firstProcessor指向。

- processRequest

简单将客户端请求包Request入队列，等待run主线程处理。代码如下：

public void processRequest(Request request) {
if (!finished) {
Request upgradeRequest = null;
try {
upgradeRequest = zks.checkUpgradeSession(request);
} catch (KeeperException ke) {
if (request.getHdr() != null) {
request.getHdr().setType(OpCode.error);
request.setTxn(new ErrorTxn(ke.code().intValue()));
}
request.setException(ke);
LOG.info("Error creating upgrade request",  ke);
} catch (IOException ie) {
LOG.error("Unexpected error in upgrade", ie);
}
if (upgradeRequest != null) {
queuedRequests.add(upgradeRequest);
}
queuedRequests.add(request);
}
}

- run

主线程，从queueRequests中取出一个Request，给nextProcessor执行，从FollowerZooKeeperServer的调用链设置方法中可以看到，这个nextProcessor是CommitProcessor。然后针对写操作调用Follower的request方法（即父类Learner的request方法）处理。主要代码如下所示：

public void run() {
try {
while (!finished) {
Request request = queuedRequests.take();
if (request == Request.requestOfDeath) {
break;
}
nextProcessor.processRequest(request);
switch (request.type) {
case OpCode.sync:
zks.pendingSyncs.add(request);
zks.getFollower().request(request);
break;
case OpCode.create:
case OpCode.create2:
case OpCode.createTTL:
case OpCode.createContainer:
case OpCode.delete:
case OpCode.deleteContainer:
case OpCode.setData:
case OpCode.reconfig:
case OpCode.setACL:
case OpCode.multi:
case OpCode.check:
zks.getFollower().request(request);
break;
case OpCode.createSession:
case OpCode.closeSession:
// Don't forward local sessions to the leader.
if (!request.isLocalSession()) {
zks.getFollower().request(request);
}
break;
}
}
} catch (Exception e) {
handleException(this.getName(), e);
}
LOG.info("FollowerRequestProcessor exited loop!");
}

request方法会发送REQUEST消息给Leader。Leader收到REQUEST后触发LeaderZooKeeperServer的submitLearnerRequestrequest方法执行。

5.3.3　ObserverRequestProcessor

ObserverZooKeeperServer调用链的第一个processor。

- processRequest

简单将Request入队列，等待run主线程处理。代码如下：

public void processRequest(Request request) {
if (!finished) {
Request upgradeRequest = null;
try {
upgradeRequest = zks.checkUpgradeSession(request);
} catch (KeeperException ke) {
if (request.getHdr() != null) {
request.getHdr().setType(OpCode.error);
request.setTxn(new ErrorTxn(ke.code().intValue()));
}
request.setException(ke);
LOG.info("Error creating upgrade request",  ke);
} catch (IOException ie) {
LOG.error("Unexpected error in upgrade", ie);
}
if (upgradeRequest != null) {
queuedRequests.add(upgradeRequest);
}
queuedRequests.add(request);
}
}

- run

主线程，从queueRequests中取出一个Request，给nextProcessor执行，从ObserverZooKeeperServer的调用链设置方法中可以看到，这个nextProcessor是CommitProcessor。然后针对写操作调用Observer的request方法处理。主要代码如下所示：

Request request = queuedRequests.take();
nextProcessor.processRequest(request);
switch (request.type) {
case OpCode.sync:
zks.pendingSyncs.add(request);
zks.getObserver().request(request);
break;
case OpCode.create:
case OpCode.create2:
case OpCode.createTTL:
case OpCode.createContainer:
case OpCode.delete:
case OpCode.deleteContainer:
case OpCode.setData:
case OpCode.reconfig:
case OpCode.setACL:
case OpCode.multi:
case OpCode.check:
zks.getObserver().request(request);
break;
case OpCode.createSession:
case OpCode.closeSession:
// Don't forward local sessions to the leader.
if (!request.isLocalSession()) {
zks.getObserver().request(request);
}
break;
}

5.3.4　ProposalProcessor

用于Leader，处理写操作命令，生成PROPOSAL包通知Follower同步写操作。

两段主要代码：

1. ProposalRequestProcessor（）

AckRequestProcessor ackProcessor = new AckRequestProcessor(zks.getLeader());
syncProcessor = new SyncRequestProcessor(zks, ackProcessor);

创建SyncProcessor，当生成PROPOSAL包时，首先通过SyncProcessor持久化到磁盘，然后通过AckRequestProcessor给Leader自己发送ACK包。

2. processRequest

nextProcessor.processRequest(request);
if (request.getHdr() != null) {
try {
zks.getLeader().propose(request);
} catch (XidRolloverException e) {
throw new RequestProcessorException(e.getMessage(), e);
}
syncProcessor.processRequest(request);
}

这里的nextProcessor在Leader中的定义是CommitProcessor，等待COMMIT。

然后调用Leader的proposal方法来处理Request。

最后调用SyncProcessor将PROPOSAL持久化，并给自己发ACK。

5.3.5　CommitProcessor

对Leader发送过来的Proposal进行确认，并且确保Leader发送过来的多条写事务请求按顺序依次执行，只有前一条写事务执行完毕才继续执行下一条写事务。

CommitProcessor是处理需要确认的写事务请求，读操作不需要确认会直接忽略掉。它的nextProcessor是FinalRequestProcessor，执行具体的写事务。

定义了3个队列：

1. LinkedBlockingQueue<Request> queuedRequests ;

请求队列，所有请求通过processPacket方法首先放入该队列，等待后续处理。

2. LinkedBlockingQueue<Request> committedRequests ;

提交执行队列，这里的请求是可以提交执行的。

3. HashMap<Long, LinkedList<Request>> pendingRequests ;

等待处理请求，写事务会被放入该队列直到收到COMMIT消息才被移到commitedRequests队列，等待被处理。

4. needCommit

判断Request是否需要Commit过程。只有写操作才会需要Commit过程，不需要Commit的读操作直接进入nextProcessor处理。

protected boolean needCommit(Request request) {
switch (request.type) {
case OpCode.create:
case OpCode.create2:
case OpCode.createTTL:
case OpCode.createContainer:
case OpCode.delete:
case OpCode.deleteContainer:
case OpCode.setData:
case OpCode.reconfig:
case OpCode.multi:
case OpCode.setACL:
return true;
case OpCode.sync:
return matchSyncs;
case OpCode.createSession:
case OpCode.closeSession:
return !request.isLocalSession();
default:
return false;
}
}

5. processPacket

有Request到达时调用该方法，该方法直接将Request放入请求队列queuedRequests中，然后唤醒等待线程开始处理队列中的请求。

public void processRequest(Request request) {
if (stopped) {
return;
}
if (LOG.isDebugEnabled()) {
LOG.debug("Processing request:: " + request);
}
queuedRequests.add(request);
wakeup();
}

6. commit

将写请求Request放入committedRequests队列，等待run线程送往FinalRequestProcessor进行真正的处理。

public void commit(Request request) {
if (stopped || request == null) {
return;
}
if (LOG.isDebugEnabled()) {
LOG.debug("Committing request:: " + request);
}
committedRequests.add(request);
wakeup();
}

7. run

主线程，循环处理请求队列，顺序执行。提交请求队列中的Request按顺序执行，调用nextProcessor将Request发送给FinalRequestProcessor处理。

run的主要逻辑代码概括如下：

public void run() {
try {

int requestsToProcess = 0;
boolean commitIsWaiting = false;
do {

commitIsWaiting = !committedRequests.isEmpty();
requestsToProcess =  queuedRequests.size();
// Avoid sync if we have something to do
if (requestsToProcess == 0 && !commitIsWaiting){
// Waiting for requests to process
synchronized (this) {
while (!stopped && requestsToProcess == 0
&& !commitIsWaiting) {
wait();
commitIsWaiting = !committedRequests.isEmpty();
requestsToProcess = queuedRequests.size();
}
}
}

Request request = null;
while (!stopped && requestsToProcess > 0
&& (request = queuedRequests.poll()) != null) {
requestsToProcess--;
if (needCommit(request)
|| pendingRequests.containsKey(request.sessionId)) {
// Add request to pending
LinkedList<Request> requests = pendingRequests
.get(request.sessionId);
if (requests == null) {
requests = new LinkedList<Request>();
pendingRequests.put(request.sessionId, requests);
}
requests.addLast(request);
}
else {
sendToNextProcessor(request);
}

if (!pendingRequests.isEmpty() && !committedRequests.isEmpty()){

commitIsWaiting = true;
break;
}
}

// Handle a single committed request
if (commitIsWaiting && !stopped){
waitForEmptyPool();

if (stopped){
return;
}

// Process committed head
if ((request = committedRequests.poll()) == null) {
throw new IOException("Error: committed head is null");
}

LinkedList<Request> sessionQueue = pendingRequests
.get(request.sessionId);
if (sessionQueue != null) {
// If session queue != null, then it is also not empty.
Request topPending = sessionQueue.poll();
if (request.cxid != topPending.cxid) {
LOG.error(
"Got cxid 0x"
+ Long.toHexString(request.cxid)
+ " expected 0x" + Long.toHexString(
topPending.cxid)
+ " for client session id "
+ Long.toHexString(request.sessionId));
throw new IOException("Error: unexpected cxid for"
+ "client session");
}

topPending.setHdr(request.getHdr());
topPending.setTxn(request.getTxn());
topPending.zxid = request.zxid;
request = topPending;
}

sendToNextProcessor(request);

waitForEmptyPool();
if (sessionQueue != null) {
while (!stopped && !sessionQueue.isEmpty()
&& !needCommit(sessionQueue.peek())) {
sendToNextProcessor(sessionQueue.poll());
}
// Remove empty queues
if (sessionQueue.isEmpty()) {
pendingRequests.remove(request.sessionId);
}
}
}
} while (!stoppedMainLoop);
} catch (Throwable e) {
handleException(this.getName(), e);
}
LOG.info("CommitProcessor exited loop!");
}

1. 首先等待Request的到来

2. 然后判断是否需要COMMIT，如果需要则将Request放入pendingRequests等待被commit；否则直接送给FinalRequestProcessor处理。

3. 如果pendingRequests和committedRequests都不为空，则跳出小循环准备执行committedRequests中的请求。

4. 根据committedRequests中的Request找到该Requests对应的Session，并从该Session的pendingRequests中找出第一个Request执行，这样保证Request是按照Session的顺序执行的，而不会出现后续Request反而先被执行的场景。

5. 最后再按Session中的顺序执行读操作，防止后续读操作在写操作之前被执行，从而破坏了事务的顺序性。

5.3.6　SyncRequestProcessor

将写事务请求持久化到磁盘，防止事务丢失。它的nextProcessor分成3种场景设置：

（1）Leader

这时候nextProcessor设置为AckRequestProcessor，AckRequestProcessor直接调用Leader的processAck方法。

（2）Follower

这时候nextProcessor设置为SendAckRequestProcessor，SendAckRequestProcessor向Leader发送Ack确认包。

（3）Observer

这时候nextProcessor设置为null，表示不用回复ACK给Leader。

SyncRequestProcessor类的processRequest方法定义如下：

public void processRequest(Request request) {
queuedRequests.add(request);
}

它的逻辑是将Request放入queueRequests队列，等待run主线程处理：

run线程主要逻辑：

while (true) {
Request si = null;
if (toFlush.isEmpty()) {
si = queuedRequests.take();
} else {
si = queuedRequests.poll();
if (si == null) {
flush(toFlush);
continue;
}
}
if (si == requestOfDeath) {
break;
}
if (si != null) {
if (zks.getZKDatabase().append(si)) {
logCount++;
if (logCount > (snapCount / 2 + randRoll)) {
randRoll = r.nextInt(snapCount/2);
// roll the log
zks.getZKDatabase().rollLog();
// take a snapshot
if (snapInProcess != null && snapInProcess.isAlive()) {
LOG.warn("Too busy to snap, skipping");
} else {
snapInProcess = new ZooKeeperThread("Snapshot Thread") {
public void run() {
try {
zks.takeSnapshot();
} catch(Exception e) {
LOG.warn("Unexpected exception", e);
}
}
};
snapInProcess.start();
}
logCount = 0;
}
} else if (toFlush.isEmpty()) {
if (nextProcessor != null) {
nextProcessor.processRequest(si);
if (nextProcessor instanceof Flushable) {
((Flushable)nextProcessor).flush();
}
}
continue;
}
toFlush.add(si);
if (toFlush.size() > 1000) {
flush(toFlush);
}
}

将Request写到事务日志文件中，根据条件决定是否重启新的日志滚动文件；日志文件写入完成后调用nextProcessor处理Request。

5.3.7　AckRequestProcessor

专门给Leader使用的ACK处理单元，当Follower收到Proposal时都是通过SendAckRequestProcessor发送ACK包给Leader，而Leader在发出Proposal命令时，会调用Sync过程将Proposal写到Leader的磁盘，然后Sync过程调用AckRequestProcessor直接通知Leader收到Leader自己的ACK包，直接调用Leader的processAck方法。

这里之所以设计单独一个类处理，主要是基于代码层次清晰考虑。

public void processRequest(Request request) {
QuorumPeer self = leader.self;
if(self != null)
leader.processAck(self.getId(), request.zxid, null);
else
LOG.error("Null QuorumPeer");
}

5.3.8　SendAckRequestProcessor

SyncRequestProcessor的nextProcessor是SendAckRequestProcessor。

当Follower持久化写事务请求到磁盘后，会立即调用SendAckRequestProcessor的processRequest方法给Leader发送ACK消息，等待Leader的COMMIT消息。

SendAckRequestProcessor的代码示例：

public void processRequest(Request si) {
if(si.type != OpCode.sync){
QuorumPacket qp = new QuorumPacket(Leader.ACK,
si.getHdr().getZxid(), null,   null);
try {
learner.writePacket(qp, false);
} catch (IOException e) {
}
}
}

5.3.9　toBeAppliedProcessor

最简单的一个处理单元，只是维护一个toBeApplied列表，然后每次将列表中一条记录发送到FinalRequestProcess or处理，同时删除该记录。

5.3.10　PrepRequestProcessor

主要目的是将写操作请求记录在ZooKeeperServer的outstandingChanges队列中，outstandingChanges主要用于FinalRequestProcessor的processRequest方法中，用来确保FinalRequestProcessor是按照outstandingChanges的顺序来执行Request。

名称

说明

processRequest

写请求入内部队列，等待pRequest方法的调用

pRequest

根据请求的OpCode调不同的处理，对写操作一般调用pRequest来处理

pRequest

处理写操作，最终执行addChangeRecord

addChangeRecord

写请求入ZooKeeperServer的outstandingChanges队列，表示这个队列里的请求都是写请求。

5.3.11　FinalRequestProcessor

最终的一个调用链处理单元。

该类用于执行最终的命令，通过ZkDatabase相关接口完成操作。

不管是Leader还是Follower，最后的处理单元一定是FinalRequestProcessor。

主函数是processRequest，真正处理客户端请求的代码。

主要代码可分成两部分。

第一部分是写操作，代码逻辑片段如下：

synchronized (zks.outstandingChanges) {
rc = zks.processTxn(request);
}

该方法主要是调用ZooKeeperServer的processTxn方法来处理。具体逻辑可参考ZooKeeperServer一节。

第二部分是读操作，直接通过ZKDatabase等接口实现，比如getData和getChildren两个读操作的代码逻辑如下：

case OpCode.getData: {
lastOp = "GETD";
GetDataRequest getDataRequest = new GetDataRequest();
ByteBufferInputStream.byteBuffer2Record(request.request,
getDataRequest);
DataNode n = zks.getZKDatabase().getNode(getDataRequest.getPath());
if (n == null) {
throw new KeeperException.NoNodeException();
}
Stat stat = new Stat();
byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat,
getDataRequest.getWatch() ? cnxn : null);
rsp = new GetDataResponse(b, stat);
break;
}
case OpCode.getChildren: {
lastOp = "GETC";
GetChildrenRequest getChildrenRequest = new GetChildrenRequest();
ByteBufferInputStream.byteBuffer2Record(request.request,
getChildrenRequest);
DataNode n = zks.getZKDatabase().getNode(getChildrenRequest.getPath());
if (n == null) {
throw new KeeperException.NoNodeException();
}
List<String> children = zks.getZKDatabase().getChildren(
getChildrenRequest.getPath(), null, getChildrenRequest
.getWatch() ? cnxn : null);
rsp = new GetChildrenResponse(children);
break;
}

5.4　LeaderZooKeeperServer

Leader通过启动LeaderZooKeeperServer来接收客户端请求。

5.4.1　调用链

首先看下它的处理链的定义，从源码看出LeaderZooKeeperServer的处理链顺序如下：

- LeaderRequestProcessor ：Leader调用链开始

- PrepRequestProcessor：写操作预准备，创建Request内容

- ProposalRequestProcessor：将写请求封装成PROPOSAL包广播到所有的follower

- CommitProcessor ：判断写操作是否可以提交真正执行

- toBeAppliedProcessor：将COMMIT后的Request移到，等待发送给FinalRequestProcessor

- FinalRequestProcessor：真正的执行Zookeeper命令的地方

- SyncRequestProcessor：将写操作持久化到磁盘

- AckRequestProcessor：当proposal持久化到磁盘，向本机发送ACK包

SyncRequestProcessor和AckRequestProcessor是在ProposalRequestProcessor内部被创建的，当Leader要处理PROPOSAL命令时先自己调用SyncRequestProcessor持久化，然后通过AckRequestProcessor直接告诉Leader处理ACK逻辑（不经过QuorumPacket传递）。

调用链设置源码如下：

protected void setupRequestProcessors() {
RequestProcessor finalProcessor = new FinalRequestProcessor(this);
RequestProcessor toBeAppliedProcessor = new Leader.ToBeAppliedRequestProcessor(finalProcessor, getLeader());
commitProcessor = new CommitProcessor(toBeAppliedProcessor,
Long.toString(getServerId()), false,
getZooKeeperServerListener());
commitProcessor.start();
ProposalRequestProcessor proposalProcessor = new ProposalRequestProcessor(this, commitProcessor);
proposalProcessor.initialize();
prepRequestProcessor = new PrepRequestProcessor(this, proposalProcessor);
prepRequestProcessor.start();
firstProcessor = new LeaderRequestProcessor(this, prepRequestProcessor);
setupContainerManager();
}

5.4.2　写请求

首先我们看一下写事务的数据流转。ZooKeeperServer处理写操作和读操作是不一样的，写操作需要将事务广播到整个集群，维持集群的数据一致性；读操作只在当前服务器执行，不需要广播到整个集群。

这里我们以create事务为例说明。

首先客户端连接到Leader，然后Leader发送proposal消息给集群中所有Follower，proposal消息带有本次create事务的数据。

Follower收到proposal后首先保存到磁盘防止proposal丢失，然后回复ACK给Leader。

Leader收集到足够多的ACK后再次发送COMMIT给所有Follower，同时Leader也在本地提交proposal执行，Follower收到COMMIT之后也在本地执行proposal事务。

Leader将执行结果返回给客户端。

通过以上过程集群中所有机器都会维护同一个完整的数据库，保证了数据一致性。

这里需要注意的一点是：写操作时Zk会调用Sync过程将写操作持久化到磁盘。

写命令处理流程示意见下图：

具体流程如下：

1. 首先，客户端创建Socket连接到Leader，发送create请求给LeaderZooKeeperServer。

2. LeaderZooKeeperServer中firstProcessor被触发，执行processRequest方法。

3. 调用链执行到ProposalRequestProcessor，触发Leader的proposal方法被执行。

4. proposal方法发送PROPOSAL消息给集群中所有的Follower。

5. Follower的QuorumCnxManager接收到PROPOSAL消息

6. QuorumCnxManager调用Follower的processPacket方法。

7. processPacket方法判断是PROPOSAL消息则调用FollowerZooKeeperServer的logRequest方法。

8. logRequest调用SyncRequestProcessor方法将proposal事务持久化到磁盘。

9. SyncRequestProcessor调用nextProcessor处理，通过SendAckRequestProcessor类发送ACK消息给Leader。

10. Leader端的QuorumCnxManager接收到ACK消息。

11. 调用Leader的processAck方法处理ACK消息。

12. 调用CommitProcessor处理，判断是否可以发送COMMIT给Follower，如果可以则发送COMMIT消息给Follower，同时Leader执行到最后一个处理即FinalRequestProcessor，准备结束本次请求。

13. 返回Response给客户端，结束这次请求；同时发送COMMIT消息给Follower。

14. Follower的QuorumCnxManager接收到COMMIT消息。

15. 调用Follower的processPacket方法来处理COMMIT消息。

16. 对COMMIT消息执行FollowerZooKeeperServer的commit方法。调用到CommitProcessor处理单元，Follower最终也执行FinalRequestProcessor。

上述13步时客户端就会得到响应结果。14-16步是集群间交互，不影响客户端返回结果。

5.4.3　读请求

读操作不涉及数据和状态的变更，因此不需要维护集群数据的一致性，流程相对于写操作要简单些。

具体读操作流程：

1. Request会发送到Leader的firstProcessor处理，这里是LeaderRequestProcessor

2. LeaderRequestProcessor对读操作会进入到PrepRequestProcessor

3. PrepRequestProcessor对读操作不执行事务相关处理，然后处理链同样进入ProposalRequestProcessor

4. ProposalRequestProcessor调用Leader的propose方法，propose方法会直接将读操作请求送达nextProcessor。

5. nextProcessor到达CommitProcessor。

6. CommitProcessor同样会直接将读操作请求送达toBeAppliedProcessor。

7. toBeAppliedProcessor将读操作请求直接发送到下一步FinalRequestProcessor。

8. FinalRequestProcessor在本机执行最终的create命令，调用ZkDataBase的processTxn方法，经过DataTree完成最终的节点创建。

5.4.4　submitLearnerRequest

Follower接收到客户端写请求时，会发送REQUEST消息给Leader。REQUEST是QuorumPacket类型，携带了客户端的写请求。

Leader收到REQUEST后调用submitLearnerRequest处理该写事务。

该方法就一条语句，调用PrepRequestProcessor的processRequest方法处理Request请求，跳过了LeaderRequestProcessor这个处理单元。接下来的处理过程就和LeaderZooKeeperServer收到客户端的写请求是一样的。

方法定义如下：

public void submitLearnerRequest(Request request) {
prepRequestProcessor.processRequest(request);
}

5.5　FollowerZooKeeperServer

Follower使用的ZooKeeperServer的子类，同标准的ZooKeeperServer很相似，除了更换了调用链。

5.5.1　调用链

首先看下它的调用链的定义：

FollowerZooKeeperServer的调用链的初始化代码如下：

protected void setupRequestProcessors() {
RequestProcessor finalProcessor = new FinalRequestProcessor(this);
commitProcessor = new CommitProcessor(finalProcessor,
Long.toString(getServerId()), true, getZooKeeperServerListener());
commitProcessor.start();
firstProcessor = new FollowerRequestProcessor(this, commitProcessor);
((FollowerRequestProcessor) firstProcessor).start();
syncProcessor = new SyncRequestProcessor(this,
new SendAckRequestProcessor((Learner)getFollower()));
syncProcessor.start();
}

从源码看出LeaderZooKeeperServer的调用链顺序如下：

- FollowerRequestProcessor

- commitProcessor

- FinalRequestProcessor

- SyncRequestProcessor

- SendAckRequestProcessor

5.5.2　写请求

当客户端连接的是FollowerZooKeeperServer时，写事务请求首先需要被Follower转换成REQUEST消息发给Leader，Leader收到REQUEST后从调用链的PrepRequestProcessor开始处理，后续流程就和LeaderZooKeeperServer一样了。

通过这种方式确保了不管是Follower接受请求还是Leader接受请求，最终的事务一致性处理流程都是从Leader开始发起的。

大致流程如下。

以create举例，客户端发起创建节点请求（create命令），如果客户端连接的ZK服务端是Follower，Follower的ZooKeeperServer实际上是子类FollowerZooKeeperServer，则Follower会按照下列顺序执行：

1. 首先，客户端创建Socket连接到Leader，发送create请求给FollowerZooKeeperServer。

2. FollowerZooKeeperServer中firstProcessor被触发，执行processRequest方法。

3. 调用链执行到FollowerRequestProcessor，触发Follower的request方法被执行。request方法定义在父类Learner中。

4. request方法发送REQUEST消息给Leader。

5. Leader的QuorumCnxManager接收到REQUEST消息

6. QuorumCnxManager将REQUEST放入接收队列，等待Leader中对应的LearnerHandler处理该REQUEST消息

7. LearnerHandler调用LeaderZooKeeperServer的submitLearnerRequest方法。

8. submitLearnerRequest方法调用PrepRequestProcessor类的processRequest方法进行处理，至此进入到上节分析的LeaderZooKeeperServer的处理流程。

后续处理流程参考LeaderZooKeeperServer的写请求即可。

5.5.3　读请求

读操作不涉及数据和状态的变更，因此不需要维护集群数据的一致性，流程相对于写操作要简单些。

具体读操作流程：

1. Request会发送到Follower的firstProcessor处理，这里是FollowerRequestProcessor

2. FollowerRequestProcessor首先通过nextProcessor处理该请求，这里nextProcessor是CommitProcessor，这里CommitProcessor对读操作不做处理直接将命令发送到下个处理环节（FinalRequestProcessor直接执行）。

3. CommitProcessor并不将读操作请求转发到Follower父类Learner的request方法，CommitPeocessor到此执行基本结束。

4. FinalRequestProcessor执行读操作请求，返回结果给客户端，本次客户端命令结束。

5.5.4　logRequest

Follower接收到Leader发送的PROPOSAL包时，会触发logRequest方法被调用。

logRequest中主要代码片段如下：

Request request = new Request(hdr.getClientId(), hdr.getCxid(), hdr.getType(), hdr, txn, hdr.getZxid());
if ((request.zxid & 0xffffffffL) != 0) {
pendingTxns.add(request);
}
syncProcessor.processRequest(request);

首先将PROPOSAL携带的事务添加到pendingTxns队列等待执行（等待COMMIT消息），然后通过syncProcessor将PROPOSAL持久化到磁盘，接着调用syncProcessor的nextProcessor处理单元，这里的nextProcessor是SendAckRequestProcessor。

SendAckRequestProcessor的processRequest方法回复一条Leader.ACK类型的QuorumPacket包给Leader。Leader通过LearnerHandler收到ACK包并调用Leader的processAck方法处理。当Leader收到足够的ACK后会发送COMMIT消息给Follower，此时Follower执行pendingTxns中等待被执行的proposal。

5.5.5　commit

当FollowerZooKeeperServer收到Leader发来的COMMIT包时，会触发commit方法被执行，这时Follower开始执行之前持久化到磁盘并且在pendingTxns队列中等待执行的PROPOSAL。

commit方法会将PROPOSAL请求发送给CommitProcessor，调用CommitProcessor的commit方法进行处理。最后实际执行会进入到FinalRequestProcessor执行。

相关代码片段如下：

Request request = pendingTxns.remove();
commitProcessor.commit(request);

5.6　ObServerZooKeeperServer

Observer不接受客户端请求，因此和Follower相比少了logRequest过程。其余和FollowerZooKeeperServer一样。

5.6.1　调用链

首先看下它的处理链的定义：

ObserverZooKeeperServer的调用链的初始化代码如下：

RequestProcessor finalProcessor = new FinalRequestProcessor(this);
commitProcessor = new CommitProcessor(finalProcessor,
Long.toString(getServerId()), true,
getZooKeeperServerListener());
commitProcessor.start();
firstProcessor = new ObserverRequestProcessor(this, commitProcessor);
((ObserverRequestProcessor) firstProcessor).start();
if (syncRequestProcessorEnabled) {
syncProcessor = new SyncRequestProcessor(this, null);
syncProcessor.start();
}

从源码看出LeaderZooKeeperServer的处理链顺序如下：

1. ObserverRequestProcessor

2. CommitProcessor

3. FinalRequestProcessor

5.6.2　commitRequest

Leader会将事务通过INFORM通知Observer执行。

Observer接收到INFORM包时，会调用到ObserverZooKeeperServer的commitRequest方法进行处理，从这里调用CommitProcessor的commit方法，最后到FinalRequestProcessor执行命令。

if (syncRequestProcessorEnabled) {
syncProcessor.processRequest(request);
}
commitProcessor.commit(request);

5.7　NettyServerCnxn

ZooKeeper客户端和服务端底层通信的接口，默认采用Netty组件作为Socket接口。客户端提交的命令会通过NettyServerCnxn到达服务端。我们看下其中最重要的receiveMessage方法。

当Socket接收到ZooKeeper客户端的请求时，触发该方法，调用ZooKeeperServer的processPacket方法。其中代码片段如下：

public void receiveMessage(ChannelBuffer message) {
if (initialized) {
zks.processPacket(this, bb);
if (zks.shouldThrottle(outstandingCount.incrementAndGet())) {
disableRecvNoWait();
}
} else {
LOG.debug("got conn req request from "
+ getRemoteSocketAddress());
zks.processConnectRequest(this, bb);
initialized = true;
}
}

当客户端第一次连接到服务端时，会调用ZooKeeperServer的processConnectRequest方法处理；其他情况下会调用ZooKeeperServer的processPacket方法处理客户端请求。具体的处理逻辑可参见ZooKeeperServer及其子类（LeaderZooKeeperServer、FollowerZooKeeperServer、ObserverZooKeeperServer）的相关代码。

6　Zookeeper数据库

ZooKeeper中数据的最终操作都是通过ZkDataBase来进行的。ZkDataBase通过DataTree保存整个树形数据结构。一个DataTree的每个节点是DataNode来表示。

ZkDataBase读数据和写数据的实现逻辑都是通过DataTree实现的，主要包括以下操作接口。

名称

说明

processTxn

所有写操作调用接口，只一行代码，直接调用DataTree的同名方法实现功能。

getNode

调用DataTree的方法获取节点的DataNode对象。

getData

调用DataTree的方法获取节点的关联数据。

getACL

调用DataTree的方法获取节点的ACL。

getChildren

调用DataTree的方法获取子节点名称列表。

其中getNode、getData、getACL、getChildren都是读接口，这里只列出了这4个读接口作为示例，实际上还有其他的读接口，实现思想是一样的，所有这里就没有全部列出来。

6.1　DataTree

6.1.1　概述

Zookeeper的业务数据都通过名为DataTree的树形数据结构来维护，DataTree的叶子节点为DataNode。

DataTree通过一个名为nodes的HashMap维护整个树的数据，nodes属性定义如下：

private final ConcurrentHashMap<String, DataNode> nodes;

这个HashMap的key是节点路径名称，value是DataNode对象，每个DataNode表示了该节点路径的数据和状态。

Zookeeper客户端的读请求和写请求达到DataTree时稍微有所区别，写请求又称写事务，统一通过processTxn方法来处理；而读请求则分散在各个不同的方法中被调用。

读请求处理方法主要包括以下几种：

名称

说明

getData

返回指定节点名称的关联数据，返回类型为byte[]。
该方法可指定Watcher参数，当节点关联数据变更时触发Watcher。

getACL

返回指定节点名称的ACL

getChildren

返回指定节点名称的子节点，返回类型List<String>。
该方法可指定Watcher参数，当节点的子节点列表变更时触发Watcher。

statNode

返回指定节点名称的状态，返回类型Stat。
该方法可指定Watcher参数，当节点的状态发生变化时触发Watcher。

写请求的具体实现统一在processTxn中被调用，主要则包括以下几种写请求：

名称

说明

createNode

创建节点，该方法会触发该节点的父节点的getChildren方法注册的Watcher。

deleteNode

删除节点，该方法会触发两种Watcher，一是该节点的父节点的getChildren方法注册的Watcher；二是该节点自身的getData方法注册的Watcher。

setData

设置节点关联数据，该方法会触发注册在该节点上数据变更类型的Watcher

setACL

设置节点的ACL，该方法不会触发Watcher。

下面详细DataTree中主要方法的实现逻辑。

6.1.2　serialize

序列化，将DataTree保存到磁盘。循环将数据节点序列化到磁盘，用到serializeNode方法。该方法是deserialize方法的逆过程，具体代码可直接参考源码。

6.1.3　deserialize

反序列化，从磁盘导入所有节点数据，保存到nodes中。

主要代码摘取如下：

nodes.clear();
String path = ia.readString("path");
while (!"/".equals(path)) {
DataNode node = new DataNode();
ia.readRecord(node, "node")
nodes.put(path, node);
synchronized (node) {
aclCache.addUsage(node.acl);
}
int lastSlash = path.lastIndexOf('/');
if (lastSlash == -1) {
root = node;
} else {
String parentPath = path.substring(0, lastSlash);
DataNode parent = nodes.get(parentPath);
if (parent == null) {
throw new IOException("Invalid Datatree, unable to find " +  "parent " + parentPath + " of path " + path);
}
parent.addChild(path.substring(lastSlash + 1));
long eowner = node.stat.getEphemeralOwner();
}
path = ia.readString("path");
}
nodes.put("/", root);

序列化和反序列化是磁盘文件相关操作，想要进一步了解细节可以看源码。

6.1.4　　processTxn

主要用于处理写事务。示例代码如下：

switch (header.getType()) {
case OpCode.create:
CreateTxn createTxn = (CreateTxn) txn;
rc.path = createTxn.getPath();
createNode(
createTxn.getPath(),
createTxn.getData(),
createTxn.getAcl(),
createTxn.getEphemeral() ? header.getClientId() : 0,
createTxn.getParentCVersion(),
header.getZxid(), header.getTime(), null);
break;
case OpCode.create2:
CreateTxn create2Txn = (CreateTxn) txn;
rc.path = create2Txn.getPath();
Stat stat = new Stat();
createNode(
create2Txn.getPath(),
create2Txn.getData(),
create2Txn.getAcl(),
create2Txn.getEphemeral() ? header.getClientId() : 0,
create2Txn.getParentCVersion(),
header.getZxid(), header.getTime(), stat);
rc.stat = stat;
break;
case OpCode.createTTL:
CreateTTLTxn createTtlTxn = (CreateTTLTxn) txn;
rc.path = createTtlTxn.getPath();
stat = new Stat();
createNode(
createTtlTxn.getPath(),
createTtlTxn.getData(),
createTtlTxn.getAcl(),
EphemeralType.ttlToEphemeralOwner(createTtlTxn.getTtl()),
createTtlTxn.getParentCVersion(),
header.getZxid(), header.getTime(), stat);
rc.stat = stat;
break;
case OpCode.createContainer:
CreateContainerTxn createContainerTxn = (CreateContainerTxn) txn;
rc.path = createContainerTxn.getPath();
stat = new Stat();
createNode(
createContainerTxn.getPath(),
createContainerTxn.getData(),
createContainerTxn.getAcl(),
EphemeralType.CONTAINER_EPHEMERAL_OWNER,
createContainerTxn.getParentCVersion(),
header.getZxid(), header.getTime(), stat);
rc.stat = stat;
break;
case OpCode.delete:
case OpCode.deleteContainer:
DeleteTxn deleteTxn = (DeleteTxn) txn;
rc.path = deleteTxn.getPath();
deleteNode(deleteTxn.getPath(), header.getZxid());
break;
case OpCode.reconfig:
case OpCode.setData:
SetDataTxn setDataTxn = (SetDataTxn) txn;
rc.path = setDataTxn.getPath();
rc.stat = setData(setDataTxn.getPath(), setDataTxn
.getData(), setDataTxn.getVersion(), header
.getZxid(), header.getTime());
break;
case OpCode.setACL:
SetACLTxn setACLTxn = (SetACLTxn) txn;
rc.path = setACLTxn.getPath();
rc.stat = setACL(setACLTxn.getPath(), setACLTxn.getAcl(),
setACLTxn.getVersion());
break;
case OpCode.closeSession:
killSession(header.getClientId(), header.getZxid());
break;
case OpCode.error:
ErrorTxn errTxn = (ErrorTxn) txn;
rc.err = errTxn.getErr();
break;
case OpCode.check:
CheckVersionTxn checkTxn = (CheckVersionTxn) txn;
rc.path = checkTxn.getPath();
break;
}

该方法是写事务操作的统一入口，具体实现分散在DataTree类中的其他方法中，下面就举几个实际的例子来分析它的代码逻辑。

- createNode

新建路径，参数是路径名称、关联数据和可选的Watcher，主要代码如下：

StatPersisted stat = new StatPersisted();
stat.setCtime(time);
stat.setMtime(time);
stat.setCzxid(zxid);
stat.setMzxid(zxid);
stat.setPzxid(zxid);
stat.setVersion(0);
stat.setAversion(0);
DataNode parent = nodes.get(parentName);
parent.stat.setCversion(parentCVersion);
parent.stat.setPzxid(zxid);
Long longval = aclCache.convertAcls(acl);
DataNode child = new DataNode(data, longval, stat);
parent.addChild(childName);
nodes.put(path, child);
dataWatches.triggerWatch(path, Event.EventType.NodeCreated);
childWatches.triggerWatch(parentName.equals("") ? "/" : parentName,Event.EventType.NodeChildrenChanged);

创建一个DataNode，设置DataNode的stat和data[]，将创建成功的DataNode添加到nodes中，key设置为创建节点的名称。并且将创建的DataNode添加到父节点的children中。

最后触发已注册到NodeCreated事件的Watcher。

- getData

获取节点关联数据，服务端示例代码：

public byte[] getData(String path, Stat stat, Watcher watcher) {
DataNode n = nodes.get(path);
if (n == null) {
throw new KeeperException.NoNodeException();
}
synchronized (n) {
n.copyStat(stat);
if (watcher != null) {
dataWatches.addWatch(path, watcher);
}
return n.data;
}
}

获取给定的path节点关联的数据，该方法提供一个Watcher参数，当该参数不为空时，DataTree会记录该path关联的Watcher，并且在该path数据发生变化时触发该Watcher，同样客户端的Watcher监控事件此时将会被触发，该Watcher定义的process(WatchedEvent event)接口会被调用。

- statNode

获取节点状态。主要代码如下：

Stat stat = new Stat();
DataNode n = nodes.get(path);
if (watcher != null) {
dataWatches.addWatch(path, watcher);
}
if (n == null) {
throw new KeeperException.NoNodeException();
}
synchronized (n) {
n.copyStat(stat);
return stat;
}

从nodes数组中获取path关联的DataNode属性，如果有Watcher则将warcher添加到dataWatches数组中，下次该DataNode变化时会触发dataWatches中的watcher被调用。

- getACL

获取节点ACL。主要代码如下：

DataNode n = nodes.get(path);
if (n == null) {
throw new KeeperException.NoNodeException();
}
synchronized (n) {
n.copyStat(stat);
return new ArrayList<ACL>(aclCache.convertLong(n.acl));
}

- getChildren

获取节点的子节点。主要代码如下：

DataNode n = nodes.get(path);
if (n == null) {
throw new KeeperException.NoNodeException();
}
synchronized (n) {
if (stat != null) {
n.copyStat(stat);
}
List<String> children=new ArrayList<String>(n.getChildren());
if (watcher != null) {
childWatches.addWatch(path, watcher);
}
return children;
}

- setData

更新节点关联数据，返回更新后的该节点状态。

Stat s = new Stat();
DataNode n = nodes.get(path);
if (n == null) {
throw new KeeperException.NoNodeException();
}
byte lastdata[] = null;
synchronized (n) {
lastdata = n.data;
n.data = data;
n.stat.setMtime(time);
n.stat.setMzxid(zxid);
n.stat.setVersion(version);
n.copyStat(s);
}
// now update if the path is in a quota subtree.
String lastPrefix = getMaxPrefixWithQuota(path);
if(lastPrefix != null) {
this.updateBytes(lastPrefix, (data == null ? 0 : data.length)
- (lastdata == null ? 0 : lastdata.length));
}
dataWatches.triggerWatch(path, EventType.NodeDataChanged);
return s;

重新设置DataNode的data[]内容，触发该DataNode上已经注册的观察者，触发这些观察者的process方法。

- deleteNode

删除节点，输入参数：

1. String path：被删除节点的名称。

2. long zxid：当前zxid。

主要代码如下：

int lastSlash = path.lastIndexOf('/');
String parentName = path.substring(0, lastSlash);
String childName = path.substring(lastSlash + 1);
DataNode node = nodes.get(path);
if (node == null) {
throw new KeeperException.NoNodeException();
}
nodes.remove(path);
synchronized (node) {
aclCache.removeUsage(node.acl);
}
DataNode parent = nodes.get(parentName);
if (parent == null) {
throw new KeeperException.NoNodeException();
}
synchronized (parent) {
parent.removeChild(childName);
parent.stat.setPzxid(zxid);
long eowner = node.stat.getEphemeralOwner();
EphemeralType ephemeralType = EphemeralType.get(eowner);
if (ephemeralType == EphemeralType.CONTAINER) {
containers.remove(path);
} else if (ephemeralType == EphemeralType.TTL) {
ttls.remove(path);
} else if (eowner != 0) {
HashSet<String> nodes = ephemerals.get(eowner);
if (nodes != null) {
synchronized (nodes) {
nodes.remove(path);
}
}
}
}
if (parentName.startsWith(procZookeeper) && Quotas.limitNode.equals(childName)) {
pTrie.deletePath(parentName.substring(quotaZookeeper.length()));
}
Set<Watcher> processed = dataWatches.triggerWatch(path,EventType.NodeDeleted);
childWatches.triggerWatch(path, EventType.NodeDeleted, processed);
childWatches.triggerWatch("".equals(parentName) ? "/" : parentName,
EventType.NodeChildrenChanged);

删除指定路径的节点，同时修改该节点的父节点的子节点列表，完成以上操作后再触发监听在该节点上的Watcher事件，注意这里被触发的Watcher事件包括两种，一种是该节点自身的Watcher，一种是父节点的getChildren监听的Watcher。

- setACL

设置节点ACL。主要代码如下：

Stat stat = new Stat();
DataNode n = nodes.get(path);
if (n == null) {
throw new KeeperException.NoNodeException();
}
synchronized (n) {
aclCache.removeUsage(n.acl);
n.stat.setAversion(version);
n.acl = aclCache.convertAcls(acl);
n.copyStat(stat);
return stat;
}

该操作不涉及到Watcher。

6.2　DataNode

数据节点的定义，包含了数据节点的数据、访问控制、状态、子节点列表等属性，定义如下：

public class DataNode implements Record {
byte data[];
Long acl;
public StatPersisted stat;
private Set<String> children = null;
}

7　Watcher机制

7.1　Watcher接口

Zookeeper对写操作提供了观察者模式，可以在监控对象内容发生变化时实时通知观察者，这一机制的实现是通过Watcher接口实现的。

客户端可以定义自己的Watcher实现，捕捉Zookeeper的内容变更事件。

变更类型大致分为三类，分别是子节点发生变化、节点关联数据发生变化、其他变化类型。定义如下：

public enum WatcherType {
Children(1), Data(2), Any(3);
}

Watcher的主要接口只有一个：

abstract public void process(WatchedEvent event);

Watcher接口中是两种事件类型，分别是KeeperState和EventType。

KeeperState包含ZK连接状态变化事件，包括连接、同步、断开连接、认证失败等ZK状态事件。

Disconnected

未连接

NoSyncConnected

废弃

SyncConnected

连接状态

AuthFailed

认证失败状态

ConnectedReadOnly

客户端连接后只读状态，此状态下客户端无权写数据

SaslAuthenticated

SASL认证过

Expired

过期状态

EventType包含节点创建、节点设置数据、节点删除、子节点变化等事件类型。

NodeCreated

节点创建事件，节点一般指一个ZK目录

NodeDeleted

节点删除事件

NodeDataChanged

节点的数据改变事件

NodeChildrenChanged

子节点改变状态

DataWatchRemoved

数据观察者移除事件

ChildWatchRemoved

通过Watch机制，客户端可以注册监听某个路径的变化，例如每当路径的关联数据变化时触发Watcher事件。我们以getData方法来看看Watcher的实现机制到底是怎样的。

public void testGetDataWatch() {
String result = null;
try {
byte[] bytes = zooKeeper.getData("/zk001", new Watcher() {
public void process(WatchedEvent event) {
LOGGER.info("testGetDataWatch  watch : {}", event.getType());
}
}, null);
} catch (Exception e) {
LOGGER.error(e.getMessage());
}
}

getData方法在调用的时候可以设置一个Watcher，这样当该path的数据变化时会触发该Watcher被调用。

7.2　客户端

ZooKeeper客户端源码里和getData方法相关的部分如下：

public byte[] getData(final String path, Watcher watcher, Stat stat)  {
final String clientPath = path;
PathUtils.validatePath(clientPath);
// the watch contains the un-chroot path
WatchRegistration wcb = null;
if (watcher != null) {
wcb = new DataWatchRegistration(watcher, clientPath);
}
final String serverPath = prependChroot(clientPath);
RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.getData);
GetDataRequest request = new GetDataRequest();
request.setPath(serverPath);
request.setWatch(watcher != null);
GetDataResponse response = new GetDataResponse();
ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
if (stat != null) {
DataTree.copyStat(response.getStat(), stat);
}
return response.getData();
}

对Watcher定义了WatchRegistration。

wcb = new DataWatchRegistration(watcher, clientPath);

将watcher与path关联起来，这样就知道哪个path的数据变化时会触发该Watcher。

下一条语句：

request.setWatch(watcher != null);

这条语句告诉ZooKeeperServer这个Request请求是关联了Watcher的，这样当Server端监测到该path的数据变化时会通知到ZooKeeper客户端。那么Server端是在什么时候通知客户端，又是如何通知客户端的呢？我们接着看看Server端的相关代码。

7.3　服务端

ZooKeeperServer接收到getData请求后，首先processPacket方法会被调用，我们先看看该方法中和Watcher机制相关的代码逻辑。

public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer)  {
submitRequest(si);
}

调用了submitRequest方法处理getData这个Request。再往下：

public void submitRequest(Request si)  {
firstProcessor.processRequest(si);
}

Request进入到服务端的调用链，我们假设这次是Leader作为服务端，则进入LeaderRequestProcessor并沿着调用链最终走到FinalRequestProcessor。这里我们直接找到和Watcher相关的代码逻辑。

最后我们发现在FinalRequestProcessor中和Watcher相关的代码：

public void processRequest(Request request) {
case OpCode.getData: {
lastOp = "GETD";
GetDataRequest getDataRequest = new GetDataRequest();
ByteBufferInputStream.byteBuffer2Record(request.request,
getDataRequest);
DataNode n = zks.getZKDatabase().getNode(getDataRequest.getPath());
if (n == null) {
throw new KeeperException.NoNodeException();
}
Stat stat = new Stat();
byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat,
getDataRequest.getWatch() ? cnxn : null);
rsp = new GetDataResponse(b, stat);
break;
}
}

调用到ZKDatabase的getData来处理客户端请求，在参数中也指明了是否设置了Watcher。如果客户端设置了Watcher则把cnxn作为Watcher传给ZKDatabase。Cnzx的定义为：

ServerCnxn cnxn = request.cnxn;

我们等一会再来看看这个ServerCnxn是怎么作为Watcher机制的，先看看ZKDatabase的getData方法：

public byte[] getData(String path, Stat stat, Watcher watcher){
return dataTree.getData(path, stat, watcher);
}

走到了DataTree的getData方法：

public byte[] getData(String path, Stat stat, Watcher watcher)
throws KeeperException.NoNodeException {
DataNode n = nodes.get(path);
if (n == null) {
throw new KeeperException.NoNodeException();
}
synchronized (n) {
n.copyStat(stat);
if (watcher != null) {
dataWatches.addWatch(path, watcher);
}
return n.data;
}
}

在这里我们看到将Watcher添加到了dataWatches列表中，该列表定义如下：

private final WatchManager dataWatches = new WatchManager();

7.4　WatchManager

7.4.1　addWatch方法

synchronized void addWatch(String path, Watcher watcher) {
HashSet<Watcher> list = watchTable.get(path);
if (list == null) {
list = new HashSet<Watcher>(4);
watchTable.put(path, list);
}
list.add(watcher);
HashSet<String> paths = watch2Paths.get(watcher);
if (paths == null) {
// cnxns typically have many watches, so use default cap here
paths = new HashSet<String>();
watch2Paths.put(watcher, paths);
}
paths.add(path);
}

将Watcher记录到本地列表中，Watcher和path关联起来了，那么这些Watcher在什么时候被触发呢？

7.4.2　triggerWatch方法

Watcher是通过triggerWatch方法被触发的。

Set<Watcher> triggerWatch(String path, EventType type, Set<Watcher> supress) {
WatchedEvent e = new WatchedEvent(type, KeeperState.SyncConnected, path);
HashSet<Watcher> watchers;
for (Watcher w : watchers) {
if (supress != null && supress.contains(w)) {
continue;
}
w.process(e);
}
return watchers;
}

具体在这里就是回调NettyServerCnxn的process方法，这里再提醒一下，NettyServerCnxn是实现了Watcher接口的。

Public Interface Watcher{
public abstract void process(WatchedEvent event);
}

那么triggerWatch在哪里被触发呢？在getData这个例子里，triggerWatch在setData中被触发。setData方法是设置数据的，设置数据时触发getData的Watcher。

public Stat setData(String path, byte data[], int version, long zxid,
long time) throws KeeperException.NoNodeException {
Stat s = new Stat();
DataNode n = nodes.get(path);
if (n == null) {
throw new KeeperException.NoNodeException();
}
byte lastdata[] = null;
synchronized (n) {
lastdata = n.data;
n.data = data;
n.stat.setMtime(time);
n.stat.setMzxid(zxid);
n.stat.setVersion(version);
n.copyStat(s);
}
String lastPrefix = getMaxPrefixWithQuota(path);
if(lastPrefix != null) {
this.updateBytes(lastPrefix, (data == null ? 0 : data.length)
- (lastdata == null ? 0 : lastdata.length));
}
dataWatches.triggerWatch(path, EventType.NodeDataChanged);
return s;
}

7.5　NettyServerCnxn

ServerCnxn中实现了Watcher接口，我们看下Netty实现的具体处理逻辑。

@Override
public void process(WatchedEvent event) {
ReplyHeader h = new ReplyHeader(-1, -1L, 0);
// Convert WatchedEvent to a type that can be sent over the wire
WatcherEvent e = event.getWrapper();
try {
sendResponse(h, e, "notification");
} catch (IOException e1) {
if (LOG.isDebugEnabled()) {
LOG.debug("Problem sending to " + getRemoteSocketAddress(), e1);
}
close();
}
}

sendResponse方法将WatchedEvent事件发送给客户端。客户端通过ClientCnxn来接收并向上处理这个WatchedEvent。

public void sendResponse(ReplyHeader h, Record r, String tag) {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
try {
baos.write(fourBytes);
bos.writeRecord(h, "header");
if (r != null) {
bos.writeRecord(r, tag);
}
baos.close();
} catch (IOException e) {
LOG.error("Error serializing response");
}
byte b[] = baos.toByteArray();
ByteBuffer bb = ByteBuffer.wrap(b);
bb.putInt(b.length - 4).rewind();
sendBuffer(bb);
}

最终服务端向客户端发送了notification消息，消息内容为WatchedEvent，那么客户端接收到WatchedEvent之后如何处理呢，这又要回到客户端代码部分了。

7.6　ClientCnxn

客户端收到WatchedEvent事件，在ClientCnxn的SendThread中，我们又回到readResponse方法。

void readResponse(ByteBuffer incomingBuffer) throws IOException {
if (replyHdr.getXid() == -1) {
// -1 means notification
WatcherEvent event = new WatcherEvent();
event.deserialize(bbia, "response");
if (chrootPath != null) {
String serverPath = event.getPath();
if(serverPath.compareTo(chrootPath)==0)
event.setPath("/");
else if (serverPath.length() > chrootPath.length())
event.setPath(serverPath.substring(chrootPath.length()));
else {
LOG.warn("Got server path " + event.getPath());
}
}
WatchedEvent we = new WatchedEvent(event);
eventThread.queueEvent( we );
return
}
}

将接收到的WatchedEvent事件放入eventThread线程的等待队列中，等待处理。

EventThread的主方法：

private void processEvent(Object event) {
if (event instanceof WatcherSetEventPair) {
// each watcher will process the event
WatcherSetEventPair pair = (WatcherSetEventPair) event;
for (Watcher watcher : pair.watchers) {
try {
watcher.process(pair.event);
} catch (Throwable t) {
LOG.error("Error while calling watcher ", t);
}
}
}
}

至此客户端在调用getData方法时绑定的Watcher被调用，process方法被执行。

8　事务持久化

8.1　概述

Zookeeper中主要有两类信息需要持久化。

一类是写操作命令，ZooKeeper在执行写命令前都要记录到事务日志中，防止写命令在集群中失败，提供恢复机制。

另一类就是snapshot快照，将整个Zookeeper的数据库持久化到磁盘。

8.2　接口

8.2.1　TxnLog

写事务日志接口，主要包含以下接口：

名称

说明

void rollLog()

日志滚动，启动新日志

boolean append(TxnHeader hdr, Record r)

添加写事务到日志尾部

long getLastLoggedZxid()

读取最后记录的zxid

boolean truncate(long zxid)

截断写事务，用在Learner事务比Leader事务多的场景

void commit()

提交事务，确认事务已持久化

rollLog方法提供日志滚动功能，如果任随事务日志文件无限增长，势必影响到性能，rollLog方法会重新启动新的事务日志文件名，后续事务写到新的文件中，ZooKeeper中多个事务日志文件通常会以zxid来区别。

Truncate方法提供截断日志功能，将zxid之后的事务全部删除。通常当Follower的事务日志比Leader还多的时候会触发改方法，保证Follower和Leader的数据库同步。

Commit方法提交文件缓存到磁盘，确保事务真实写到磁盘而不是仅仅存在于文件内存缓存中。

8.2.2　SnapShot

快照接口

主要是两个接口：序列化和反序列化，通过这个接口QuorumPeer可以快速的将自己的完整快照传递给远方QuorumPeer，实现数据在集群间的快速同步。

名称

说明

deserialize

从序列化文件中恢复数据库和session

serialize

将数据库和session序列化保存到文件

8.3　实现类

ZooKeeper中实现TxnLog和SnapShot接口的类是FileTxnLog和FileSnap，感兴趣的读者也可以扩展自己的实现，比如可以将事务保存到MySql或者Hbase等等，只要掌握了源码的核心设计模式做到这些并不困难。

8.3.1　FileTxnLog

写事务日志实现类。

- append方法

主要代码：

if (logStream==null) {
logFileWrite = new File(logDir, ("log." + Long.toHexString(hdr.getZxid())));
fos = new FileOutputStream(logFileWrite);
logStream=new BufferedOutputStream(fos);
oa = BinaryOutputArchive.getArchive(logStream);
FileHeader fhdr = new FileHeader(TXNLOG_MAGIC,VERSION, dbId);
fhdr.serialize(oa, "fileheader");
logStream.flush();
currentSize = fos.getChannel().position();
streamsToFlush.add(fos);
}
padFile(fos);
Checksum crc = makeChecksumAlgorithm();
crc.update(buf, 0, buf.length);
oa.writeLong(crc.getValue(), "txnEntryCRC");
Util.writeTxnBytes(oa, buf);

该方法将事务添加到日志文件的尾部。

日志没滚动前，写到当前日志文件；日志回滚的话写到新的日志文件，新日志文件的名称是”log.”加当前zxid的值。

- rollLog方法

日志滚动，关闭旧的日志文件，启动新的日志文件，主要代码：

if (logStream != null) {
this.logStream.flush();
this.logStream = null;
oa = null;
}

- getLastLoggedZxid方法

从日志文件中获取最近记录的zxid的值，从lastLoggedZxid就能获得最新的事务日志文件。多个日志文件的情况会遍历所有文件，选出文件名中zxid最大的那个日志文件，再从该日志文件中取到最大的zxid。

主要代码：

public long getLastLoggedZxid() {
File[] files = getLogFiles(logDir.listFiles(), 0);
long maxLog=files.length>0?
Util.getZxidFromName(files[files.length-1].getName(),"log"):-1;
long zxid = maxLog;
TxnIterator itr = null;
try {
FileTxnLog txn = new FileTxnLog(logDir);
itr = txn.read(maxLog);
while (true) {
if(!itr.next())
break;
TxnHeader hdr = itr.getHeader();
zxid = hdr.getZxid();
}
} catch (IOException e) {
LOG.warn("Unexpected exception", e);
} finally {
close(itr);
}
return zxid;
}

- truncate方法

截断多余的日志信息，保证日志文件是合法有效的。

主要代码：

public boolean truncate(long zxid) throws IOException {
FileTxnIterator itr = null;
try {
itr = new FileTxnIterator(this.logDir, zxid);
PositionInputStream input = itr.inputStream;
long pos = input.getPosition();
RandomAccessFile raf=new RandomAccessFile(itr.logFile,"rw");
raf.setLength(pos);
raf.close();
while(itr.goToNextLog()) {
if (!itr.logFile.delete()) {
LOG.warn("Unable to truncate {}", itr.logFile);
}
}
} finally {
close(itr);
}
return true;
}

- commit方法

确认提交日志文件缓存，主要代码：

public synchronized void commit() throws IOException {
if (logStream != null) {
logStream.flush();
}
for (FileOutputStream log : streamsToFlush) {
log.flush();
if (forceSync) {
long startSyncNS = System.nanoTime();
log.getChannel().force(false);
long syncElapsedMS = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startSyncNS);
}
}
while (streamsToFlush.size() > 1) {
streamsToFlush.removeFirst().close();
}
}

8.3.2　FileSnap

SnapShot接口的实现类。主要实现deserialize和serialize两个接口，具体代码参考源码。这里就不花篇幅详细介绍了，搞清楚FileTxnLog再理解FileSnap就很容易了。

9　命令行

9.1　概述

前面分析了源码的结构和流程，对ZooKeeper的内部运行机制大致有了了解。这节看看ZooKeeper是怎么从命令行一步步启动到QuorumPeer的。

命令行的执行脚本在bin目录，而命令行的Java实现在cli目录。

9.2　bin目录

脚本列表和说明：

名称

说明

zkEnv.sh

设置ZK服务器运行时环境变量，ZOOCFGDIR、ZOO_LOG_DIR等

zkServer-initialize.sh

服务器初始化，设置myid

zkServer.sh

启动Server
首先运行zkEnv.sh
配置JMX
Kill还在运行的zk进程
判断data_dir/log_dir是否存在
根据命令行参数启动server，支持start/restart/start-foreground/status等命令方式。
运行"org.apache.zookeeper.server.quorum.QuorumPeerMain"类

zkCli.sh

启动客户端，运行：
org.apache.zookeeper.ZooKeeperMain

9.2.1　ZkServer.cmd

服务端启动脚本，下面我们看看Windows平台下服务端启动代码如何调用到Java类的。

setlocal
call "%~dp0zkEnv.cmd"

set ZOOMAIN=org.apache.zookeeper.server.quorum.QuorumPeerMain
set ZOO_LOG_FILE=zookeeper-%USERNAME%-server-%COMPUTERNAME%.log

echo on
call %JAVA% "-Dzookeeper.log.dir=%ZOO_LOG_DIR%" "-Dzookeeper.root.logger=%ZOO_LOG4J_PROP%" "-Dzookeeper.log.file=%ZOO_LOG_FILE%" "-XX:+HeapDumpOnOutOfMemoryError" "-XX:OnOutOfMemoryError=cmd /c taskkill /pid %%%%p /t /f" -cp "%CLASSPATH%" %ZOOMAIN% "%ZOOCFG%" %*

endlocal

从脚本中看到最终是启动QuorumPeerMain类。

9.2.2　ZKCli.cmd

客户端脚本，下面看看它是如何调用到ZooKeeper类的。

setlocal
call "%~dp0zkEnv.cmd"

ZOO_LOG_FILE=zookeeper-%USERNAME%-cli-%COMPUTERNAME%.log

set ZOOMAIN=org.apache.zookeeper.ZooKeeperMain
call %JAVA% "-Dzookeeper.log.dir=%ZOO_LOG_DIR%" "-Dzookeeper.root.logger=%ZOO_LOG4J_PROP%" "-Dzookeeper.log.file=%ZOO_LOG_FILE%" -cp "%CLASSPATH%" %ZOOMAIN% %*

endlocal

从脚本中看到最终是启动ZooKeeperMain类。

9.3　ZooKeeperMain

客户端启动类ZooKeeperMain。

Shell脚本zkCli.sh会启动ZooKeeper的客户端用户交换命令行，循环等待用户输入命令，执行用户命令。zkCli.sh会启动ZooKeeperMain类。

启动方法main

public static void main(String args[]) throws CliException, IOException, InterruptedException {
ZooKeeperMain main = new ZooKeeperMain(args);
main.run();
}

9.3.1　Run方法

在run方法中是个循环体，不断读取用户的input命令，然后执行，主要代码片段如下：

Class<?> consoleC = Class.forName("jline.console.ConsoleReader");
Class<?> completorC = Class.forName("org.apache.zookeeper.JLineZNodeCompleter");

System.out.println("JLine support is enabled");
Object console = consoleC.getConstructor().newInstance();

Object completor =completorC.getConstructor(ZooKeeper.class).newInstance(zk);
Method addCompletor = consoleC.getMethod("addCompleter",
Class.forName("jline.console.completer.Completer"));
addCompletor.invoke(console, completor);

String line;
Method readLine = consoleC.getMethod("readLine", String.class);
while ((line = (String)readLine.invoke(console, getPrompt())) != null) {
executeLine(line);
}

9.3.2　executeLine

调用processCmd方法执行客户端命令行输入。

public void executeLine(String line) throws CliException, InterruptedException, IOException {
if (!line.equals("")) {
cl.parseCommand(line);
addToHistory(commandCount,line);
processCmd(cl);
commandCount++;
}
}

9.3.3　processCmd

调用processZKCmd。

protected boolean processCmd(MyCommandOptions co) throws CliException, IOException, InterruptedException {
boolean watch = false;
try {
watch = processZKCmd(co);
exitCode = 0;
} catch (CliException ex) {
exitCode = ex.getExitCode();
System.err.println(ex.getMessage());
}
return watch;
}

9.3.4　processZKCmd

具体执行命令的入口，代码如下所示：

protected boolean processZKCmd(MyCommandOptions co) throws CliException, IOException, InterruptedException {
String[] args = co.getArgArray();
String cmd = co.getCommand();
if (args.length < 1) {
usage();
throw new MalformedCommandException("No command entered");
}
if (!commandMap.containsKey(cmd)) {
usage();
throw new CommandNotFoundException("Command not found " + cmd);
}
boolean watch = false;
LOG.debug("Processing " + cmd);
if (cmd.equals("quit")) {
zk.close();
System.exit(exitCode);
} else if (cmd.equals("redo") && args.length >= 2) {
Integer i = Integer.decode(args[1]);
if (commandCount <= i || i < 0) {
throw new MalformedCommandException("Command index out of range");
}
cl.parseCommand(history.get(i));
if (cl.getCommand().equals("redo")) {
throw new MalformedCommandException("No redoing redos");
}
history.put(commandCount, history.get(i));
processCmd(cl);
} else if (cmd.equals("history")) {
for (int i = commandCount - 10; i <= commandCount; ++i) {
if (i < 0) continue;
System.out.println(i + " - " + history.get(i));
}
} else if (cmd.equals("printwatches")) {
if (args.length == 1) {
System.out.println("printwatches is " + (printWatches ? "on" : "off"));
} else {
printWatches = args[1].equals("on");
}
} else if (cmd.equals("connect")) {
if (args.length >= 2) {
connectToZK(args[1]);
} else {
connectToZK(host);
}
}
if (zk == null || !zk.getState().isAlive()) {
System.out.println("Not connected");
return false;
}
CliCommand cliCmd = commandMapCli.get(cmd);
if(cliCmd != null) {
cliCmd.setZk(zk);
watch = cliCmd.parse(args).exec();
} else if (!commandMap.containsKey(cmd)) {
usage();
}
return watch;
}

从这里可以看到，最终是映射到CliCommand接口来执行具体的命令的。CliCommand接口针对每个具体的命令都有实现类，它们的位置在/cli目录下。

客户端支持的主要命令如下表所示。

CliCommand实现类

说明

CreateCommand

执行create命令

DeleteCommand

执行delete命令

GetAclCommand

获取DataNode的Acl

GetCommand

获取DataNode的数据

SetCommand

设置DataNode的数据

StatCommand

获取DataNode的状态属性

LsCommand

获取DataNode的子路径列表

SyncCommand

持久化DataNode到磁盘

9.4　cli目录

客户端命令行相关的类实现。主要是调用ZooKeeper类的接口。这里我们举几个例子来说明一下实现逻辑。

9.4.1　CliCommand

命令行基类，执行入口，主函数是exec，执行具体的命令。

主要接口：

abstract public boolean exec() throws CliException;

CliCommand的子类很多，都是具体命令的实现类。列表说明如下：

命令名称

说明

GetCommand

获取数据

SetCommand

设置数据

9.4.2　GetCommand

获取Zookeeper节点关联数据，实际是调用ZooKeeper的getData方法实现的。

前面说过，ZooKeeper是客户端的主要实现类，向ZooKeeperServer发送命令并获取结果。

@Override
public boolean exec() throws CliException {
boolean watch = cl.hasOption("w");
String path = args[1];
Stat stat = new Stat();
byte data[];
try {
data = zk.getData(path, watch, stat);
} catch (KeeperException|InterruptedException ex) {
throw new CliException(ex);
}
data = (data == null) ? "null".getBytes() : data;
out.println(new String(data));
if (cl.hasOption("s")) {
new StatPrinter(out).print(stat);
}
return watch;
}

9.4.3　SetCommand

设置Zookeeper节点关联数据，调用ZooKeeper的getData方法实现的。

@Override
public boolean exec() throws CliException {
String path = args[1];
byte[] data = args[2].getBytes();
int version;
if (cl.hasOption("v")) {
version = Integer.parseInt(cl.getOptionValue("v"));
} else {
version = -1;
}
try {
Stat stat = zk.setData(path, data, version);
if (cl.hasOption("s")) {
new StatPrinter(out).print(stat);
}
} catch (KeeperException|InterruptedException ex) {
throw new CliWrapperException(ex);
}
return false;
}

9.4.4　StatCommand

获取DataNode节点状态。

代码如下：

public boolean exec() throws CliException {
String path = args[1];
boolean watch = cl.hasOption("w");
Stat stat;
try {
stat = zk.exists(path, watch);
} catch (KeeperException|InterruptedException ex) {
throw new CliWrapperException(ex);
}
if (stat == null) {
throw new CliWrapperException(new
KeeperException.NoNodeException(path));
}
new StatPrinter(out).print(stat);
return watch;
}

9.4.5　LsCommand

“ls”命令，获取Zookeeper节点子节点树并通过printChildren方法打印在控制台。

代码片段：

public boolean exec() throws CliException {
if (args.length < 2) {
throw new MalformedCommandException(getUsageStr());
}
String path = args[1];
boolean watch = cl.hasOption("w");
boolean withStat = cl.hasOption("s");
boolean recursive = cl.hasOption("R");
try {
if (recursive) {
ZKUtil.visitSubTreeDFS(zk, path, watch, new StringCallback() {
@Override
public void processResult(int rc, String path, Object ctx, String name) {
out.println(path);
}
});
} else {
Stat stat = withStat ? new Stat() : null;
List<String> children = zk.getChildren(path, watch, stat);
printChildren(children, stat);
}
} catch (KeeperException|InterruptedException ex) {
throw new CliWrapperException(ex);
}
return watch;
}

以上举了几个客户端命令的代码作为例子，来理解Zookeeper中命令从控制台到java类的实现路径和执行过程，还有其他如create、reconfig等命令，流程都是相似的，这里就不再赘述了。

10　配置和应用

10.1　管理端

Zookeeper提供了web管理端来管理Zookeeper服务器，相关实现类在admin目录中。

Zookeeper服务器启动后，可通过http://<hostname>:8080/commands来得到所有可操作的命令列表，通过http://<hostname>:8080/commands/<commandname>来执行具体的命令，例如：

set_trace_mask命令：

可通过http://localhost:8080/commands/set_trace_mask?traceMask=306来执行。

返回的结果直接打印在浏览器屏幕上，Zookeeper默认使用轻量级的Jetty作为内嵌的Web服务器容器。

10.1.1　启动

通过AdminServer启动管理Web端，启动代码：

Class<?> jettyAdminServerC = Class.forName("org.apache.zookeeper.server.admin.JettyAdminServer");
Object adminServer = jettyAdminServerC.getConstructor().newInstance()
return (AdminServer) adminServer;

10.1.2　JettyAdminServer

轻量级的Jetty内核的Web管理端。

- 构造函数

绑定Web管理端地址和端口，设置servlet处理类，这里http最终是通过CommandServlet类的doGet方法来响应的。

public JettyAdminServer(String address, int port, int timeout, String commandUrl) {
this.port = port;
this.idleTimeout = timeout;
this.commandUrl = commandUrl;
this.address = address;
server = new Server();
ServerConnector connector = new ServerConnector(server);
connector.setHost(address);
connector.setPort(port);
connector.setIdleTimeout(idleTimeout);
server.addConnector(connector);
ServletContextHandler context = new ServletContextHandler(ServletContextHandler.SESSIONS);
context.setContextPath("/*");
server.setHandler(context);
context.addServlet(new ServletHolder(new CommandServlet()), commandUrl + "/*");
}

- Start

启动Jetty服务，启动org.eclipse.jetty.server.Server。

- Stop

停止Jetty服务。

10.1.3　CommandServlet

Jetty的servlet处理类。

主方法为doGet，代码片段：

// Run the command
CommandResponse cmdResponse = Commands.runCommand(cmd, zkServer, kwargs);
CommandOutputter outputter = new JsonOutputter();
response.setStatus(HttpServletResponse.SC_OK);
response.setContentType(outputter.getContentType());
outputter.output(cmdResponse, response.getWriter());

读取Web Url请求传入的命令，调用Commands.runCommand方法执行命令，将结果输出到Jetty的web界面。

10.1.4　Commands

管理端命令接口，下面举几个Command例子的代码作为示例，其他的Command可以举一反三，很容易理解。

- EnvCommand

获取环境变量列表。其中主要执行代码如下。

@Override
public CommandResponse run(ZooKeeperServer zkServer, Map<String, String> kwargs) {
CommandResponse response = initializeResponse();
for (Entry e : Environment.list()) {
response.put(e.getKey(), e.getValue());
}
return response;
}

- SrvrCommand

获取服务端的主要信息：

@Override
public CommandResponse run(ZooKeeperServer zkServer, Map<String, String> kwargs) {
CommandResponse response = initializeResponse();
LOG.info("running stat");
response.put("version", Version.getFullVersion());
response.put("read_only", zkServer instanceof ReadOnlyZooKeeperServer);
response.put("server_stats", zkServer.serverStats());
response.put("node_count", zkServer.getZKDatabase().getNodeCount());
return response;
}

- WatchCommand

查询系统有哪些Watcher，返回Map数据结构，"session_id_to_watched_paths": Map<Long, Set<String>> session ID -> Watch的路径

@Override
public CommandResponse run(ZooKeeperServer zkServer, Map<String, String> kwargs) {
DataTree dt = zkServer.getZKDatabase().getDataTree();
CommandResponse response = initializeResponse();
response.put("session_id_to_watched_paths", dt.getWatches().toMap());
return response;
}

10.2　服务器配置

配置参数详解（主要是conf/zoo.cfg文件）。

参数名

说明

clientPort

客户端连接server的端口，即对外服务端口，一般设置为2181吧。

dataDir

存储快照文件snapshot的目录。默认情况下，事务日志也会存储在这里。建议同时配置参数dataLogDir，事务日志的写性能直接影响zk性能。

tickTime

ZK中的一个时间单元。ZK中所有时间都是以这个时间单元为基础，进行整数倍配置的。例如，session的最小超时时间是2*tickTime。

dataLogDir

事务日志输出目录。尽量给事务日志的输出配置单独的磁盘或是挂载点，这将极大的提升ZK性能。

globalOutstandingLimit

最大请求堆积数。默认是1000。ZK运行的时候，尽管server已经没有空闲来处理更多的客户端请求了，但是还是允许客户端将请求提交到服务器上来，以提高吞吐性能。当然，为了防止Server内存溢出，这个请求堆积数还是需要限制下的。(Java system property:zookeeper.globalOutstandingLimit.)

preAllocSize

预先开辟磁盘空间，用于后续写入事务日志。默认是64M，每个事务日志大小就是64M。如果ZK的快照频率较大的话，建议适当减小这个参数。(Java system property:zookeeper.preAllocSize)

snapCount

每进行snapCount次事务日志输出后，触发一次快照(snapshot)，此时，ZK会生成一个snapshot.文件，同时创建一个新的事务日志文件log.。默认是100000.（真正的代码实现中，会进行一定的随机数处理，以避免所有服务器在同一时间进行快照而影响性能）(Java system property:zookeeper.snapCount)

traceFile

用于记录所有请求的log，一般调试过程中可以使用，但是生产环境不建议使用，会严重影响性能。(Java system property:? requestTraceFile)

maxClientCnxns

单个客户端与单台服务器之间的连接数的限制，是ip级别的，默认是60，如果设置为0，那么表明不作任何限制。请注意这个限制的使用范围，仅仅是单台客户端机器与单台ZK服务器之间的连接数限制，不是针对指定客户端IP，也不是ZK集群的连接数限制，也不是单台ZK对所有客户端的连接数限制。指定客户端IP的限制策略，这里有一个patch，可以尝试一下：http://rdc.taobao.com/team/jm/archives/1334（No Java system property）

clientPortAddress

对于多网卡的机器，可以为每个IP指定不同的监听端口。默认情况是所有IP都监听 clientPort指定的端口。 New in 3.3.0

minSessionTimeoutmaxSessionTimeout

Session超时时间限制，如果客户端设置的超时时间不在这个范围，那么会被强制设置为最大或最小时间。默认的Session超时时间是在2 tickTime ~ 20 tickTime 这个范围 New in 3.3.0

fsync.warningthresholdms

事务日志输出时，如果调用fsync方法超过指定的超时时间，那么会在日志中输出警告信息。默认是1000ms。(Java system property: fsync.warningthresholdms)New in 3.3.4

autopurge.purgeInterval

在上文中已经提到，3.4.0及之后版本，ZK提供了自动清理事务日志和快照文件的功能，这个参数指定了清理频率，单位是小时，需要配置一个1或更大的整数，默认是0，表示不开启自动清理功能。(No Java system property) New in 3.4.0

autopurge.snapRetainCount

这个参数和上面的参数搭配使用，这个参数指定了需要保留的文件数目。默认是保留3个。(No Java system property) New in 3.4.0

electionAlg

在之前的版本中，这个参数配置是允许我们选择leader选举算法，但是由于在以后的版本中，只会留下一种“TCP-based version of fast leader election”算法，所以这个参数目前看来没有用了，这里也不详细展开说了。(No Java system property)

initLimit

Follower在启动过程中，会从Leader同步所有最新数据，然后确定自己能够对外服务的起始状态。Leader允许F在initLimit时间内完成这个工作。通常情况下，我们不用太在意这个参数的设置。如果ZK集群的数据量确实很大了，F在启动的时候，从Leader上同步数据的时间也会相应变长，因此在这种情况下，有必要适当调大这个参数了。(No Java system property)

syncLimit

在运行过程中，Leader负责与ZK集群中所有机器进行通信，例如通过一些心跳检测机制，来检测机器的存活状态。如果L发出心跳包在syncLimit之后，还没有从F那里收到响应，那么就认为这个F已经不在线了。注意：不要把这个参数设置得过大，否则可能会掩盖一些问题。(No Java system property)

leaderServes

默认情况下，Leader是会接受客户端连接，并提供正常的读写服务。但是，如果你想让Leader专注于集群中机器的协调，那么可以将这个参数设置为no，这样一来，会大大提高写操作的性能。(Java system property: zookeeper.leaderServes)。

server.x=[hostname]:nnnnn[:nnnnn]

这里的x是一个数字，与myid文件中的id是一致的。右边可以配置两个端口，第一个端口用于F和L之间的数据同步和其它通信，第二个端口用于Leader选举过程中投票通信。(No Java system property)

group.x=nnnnn[:nnnnn]weight.x=nnnnn

对机器分组和权重设置。

cnxTimeout

Leader选举过程中，打开一次连接的超时时间，默认是5s。(Java system property: zookeeper. cnxTimeout)

zookeeper.DigestAuthenticationProvider.superDigest

ZK权限设置相关，具体可参考官网

skipACL

对所有客户端请求都不作ACL检查。如果之前节点上设置有权限限制，一旦服务器上打开这个开头，那么也将失效。(Java system property: zookeeper.skipACL)

forceSync

这个参数确定了是否需要在事务日志提交的时候调用FileChannel.force来保证数据完全同步到磁盘。(Java system property: zookeeper.forceSync)

jute.maxbuffer

每个节点最大数据量，是默认是1M。这个限制必须在server和client端都进行设置才会生效。(Java system property: jute.maxbuffer)

10.3　ZooKeeper在大数据中的应用

Zookeeper被广泛应用于各种分布式集群场景中，比如Hadoop、Storm、Spark、Kafka等，要想学好大数据分析，不可避免的要用到大规模机器集群，了解和掌握Zookeeper的工作原理将会对学好大数据分析起到事半功倍的作用。

10.3.1　实时配置文件

Zookeeper可以作为互联网应用的实时配置开关，将配置信息设置在Zookeeper的node上，并在应用中设置Zookeeper的观察者，实时获取该node关联数据的变化，达到动态实时控制应用的配置信息的目的，实时调控生产环境各项参数。

在大型电商网站中，这种方式可用于节假日等流量爆发的场景，关闭某些不太重要的电商网站功能，降低并发量大爆发对服务器性能的冲击。

10.3.2　Kafka中使用

Kafka通过Zookeeper的使用提供给客户端负载均衡能力，每个Kafka客户端直接通过连接Zookeeper就能获得整个Kafka的配置信息。

Kafka将队列服务的机器注册在Zookeeper上，同时维护客户端列表，依靠负载均衡算法将客户端动态分配到合适的队列分区上，维持整个集群的负载均衡。客户端不需要预先知道当前的队列到底分配在哪个机器上，通过Zookeeper获取队列分配的动态变化。

10.3.3　Hadoop使用

在Hadoop1.x版本中Master是单节点模式，存在单点故障不支持热备份。Hadoop2.x版本已经开始支持使用Zookeeper作为Master节点的高可靠性方案。

Hadoop客户端连接集群时指定Zookeeper服务器地址，从Zookeeper获取当前Master节点地址，进而连接Master操作Hadoop。Hadoop客户端同时监听Zookeeper中数据变化，当Master节点切换时会将切换状态变更保存到Zookeeper中，同时Zookeeper会触发数据变化监听客户端，通知Hadoop客户端切换Master地址，达到热切换（HA）的目的。

10.4　参考

ZooKeeper官网：http://zookeeper.apache.org/

Netty官网：http://netty.io/

看完了

如果您对本书内容有疑问，可发邮件至contact@turingbook.com，会有编辑或作译者协助答疑。也可访问图灵社区，参与本书讨论。

如果是有关电子书的建议或问题，请联系专用客服邮箱：ebook@turingbook.com。

在这里可以找到我们：

- 微博 @图灵教育 : 好书、活动每日播报

- 微博 @图灵社区 : 电子书和好文章的消息

- 微博 @图灵新知 : 图灵教育的科普小组

- 微信 图灵访谈 : ituring_interview，讲述码农精彩人生

- 微信 图灵教育 : turingbooks

图灵社区会员 A_Q（961353968@qq.com） 专享 尊重版权