好久没有写文章了，其实一直想写关于比特币技术的文章，确实也写了一篇，但感觉没有把问题说清楚，还需要再研究一段时间就没有发出来。
最近的工作跟zookeeper关系很近，很早就听说过zookeeper了，多多少少也接触过，但从来没读过源码，这次好好总结一下。
注；本文分析的是zookeeper原生的源码，而不是我们常用的curator，这点要搞清楚，关于curator后面也打算写篇文章总结一下，这次先来看看原生zookeeper客户端的实现原理。
一、执行过程概述
在看源码之前先梳理一下整个客户端大概的执行流程以及重要组件，搞清楚这个再看源码会稍微容易一些，这里上一张不是很准确的流程图（之所以说不是很准确是因为不是每个请求都会走这套流程）：
zk客户端流程图
简单解释一下上面的流程图：
首选在ZK中把客户端对象全部封装在类ZooKeeper中，在该类中，又把跟服务端的通信逻辑封装在类ClientCnxn中。
在一次网络通信中关键点就是三个队列：
- outgoingQueue，该队列用于存放等待发送的数据包。
- pendingQueue，该队列用于存放已发送，等待服务端响应的数据包。
- waitingEvents，服务端的响应会封装成事件放入该队列中等待处理。
注：outgoingQueue跟pendingQueue队列其实是类ClientCnxn的成员变量，我画在sendThread类中是为了好理解。
整个通信过程中，请求都按图中的流程进行。例如用户发送一个读取数据节点的请求，该请求首先被封装成一个相应类型的数据包放入outgoingQueue队列中，sendThread会把数据包从队列中取出然后发往服务端并把该数据包加入到pendingQueue中，当sendThread收到服务端响应时，会与pendingQueue队列中的第一个数据包进行对比，判断是否是期待的那个响应包（这里可以看出数据包是串行处理的），最后把响应的数据包封装成对应类型的事件加入到waitingEvents中，再由eventThread取出进行处理。
二、源码分析
由于zk的处理过程中，很大一部分都是基于事件模型的，因此在看其他源码之前，我们先总结一下zk中的事件类型，我们已经知道zk有个重要特性就是通过watcher来接收事件进行处理，这里就先贴一下watcher的源码：
public interface Watcher {
abstract public void process(WatchedEvent event);
}
Watcher是一个接口，只有一个process方法，接收WatchedEvent事件，从WatchedEvent的源码入手来看看事件的内容：
public class WatchedEvent {
final private KeeperState keeperState;
final private EventType eventType;
private String path;
//省略掉了get，set等辅助方法
}
代码很简单，只有三个成员变量，先看看连接状态的种类：
public enum KeeperState {
//表示连接断开
Disconnected (0),
//表示与服务端连接中
SyncConnected (3),
//表示连接校验失败
AuthFailed (4),
//ZK从3.3.0之后开始支持只读模式，我们都知道ZK是基于多数派的，只有
//整个集群多数节点存在才能运行，但是为了满足某些场景希望ZK集群只剩
//少数节点时依然具有读取功能，因此推出了这个状态，首先客户端需要设置
//允许只读模式，然后再连接服务端的时候，如果此时服务端只有少数节点存活（也可能是脑裂）
//则会返回这个状态给客户端，通知客户端目前只能处理读请求
ConnectedReadOnly (5),
//表示服务器采用SASL做校验
SaslAuthenticated(6),
//会话过期，连接需要重新建立，即重新初始化ZooKeeper对象
Expired (-112),
//客户端主动关闭后的状态
Closed (7);
//省略部分代码...
}
再次强调上面的ConnectedReadOnly状态，这个网上的资料很少，但却是一个非常重要的功能，以后再也不能说ZK不支持少数节点存在时处理读请求了。
节点的事件有以下几种，看名字就很好理解，就不多解释了。
public enum EventType {
None (-1),
NodeCreated (1),
NodeDeleted (2),
NodeDataChanged (3),
NodeChildrenChanged (4),
DataWatchRemoved (5),
ChildWatchRemoved (6);
//省略部分代码...
}
这里要总结一下zookeeper关于事件的分类：
zk把事件分为两大类，一个是连接状态事件KeeperState，一个是节点状态事件EventType。我们一般会认为这两类事件应该分开处理，但在zk中把他们合并了，也就是说不管是连接状态发生改变，还是节点状态发生改变，zk都会包装成一个统一的包含这两种状态的WatchedEvent类型对象。如果是连接状态发生变化，那么eventType的值永远是None，如果是节点状态发生变化，那么keeperState状态永远是SyncConnected，这点要参考上面的Watcher源码。
搞明白事件类型以后，看下客户端的入口类ZooKeeper，从构造函数开始：
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
throws IOException
{
this(connectString, sessionTimeout, watcher, false);
}
上面这段是zookeeper中最简单的构造函数，一个zookeeper对象代表了一个客户端，一个跟ZK集群的连接。用惯了curator，刚看到这个构造函数就有点蒙，watcher是必传的。
看一下最终zk构造函数的落脚点：
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
boolean canBeReadOnly, HostProvider aHostProvider,
ZKClientConfig clientConfig) throws IOException {
//如果配置为null，则初始化一个默认配置
if (clientConfig == null) {
clientConfig = new ZKClientConfig();
}
this.clientConfig = clientConfig;
//这里初始化了watchManager，顾名思义这是用来管理watcher的
//该方法对应返回的类名是ZKWatchManager，后面细说
watchManager = defaultWatchManager();
//这里我们传入的watcher对象成为了watchManager中的一个成员变量
watchManager.defaultWatcher = watcher;
//ConnectStringParser就是用来解析我们传入的zk地址字符串
ConnectStringParser connectStringParser = new ConnectStringParser(
connectString);
//用于获取zk地址
hostProvider = aHostProvider;
//初始化连接对象，逻辑主要在该类中
cnxn = new ClientCnxn(connectStringParser.getChrootPath(),
hostProvider, sessionTimeout, this, watchManager,
getClientCnxnSocket(), canBeReadOnly);
//启动客户端
cnxn.start();
}
先说一下上面代码片段中的两个对象connectStringParser跟hostProvider，zk地址的写法比较多，可以传入多个地址用都好分开，比如1.1.1.1:2181,2.2.2.2:2181也可以带上初始路径，即1.1.1.1:2181,2.2.2.2:2181/test，connectStringParser类就是用来解析我们传入的zk地址，而hostProvider类则用于负载均衡，比如我们传入三个zk地址，那么要选哪个地址进行连接？如果连接失败了要换成哪个地址进行重试？这些逻辑就在该类中，比较清楚，这里就不贴源码了。
除了上面两个辅助类，剩下的两个重点就是用于管理watch的watchManager跟用于网络通信的ClientCnxn，一个一个来看。
由于watchManager跟网络通信有很大关系，所以我们先看下ClientCnxn类的构造方法：
public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,
ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,
long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) {
//初始化一系列变量
this.zooKeeper = zooKeeper;
this.watcher = watcher;
this.sessionId = sessionId;
this.sessionPasswd = sessionPasswd;
this.sessionTimeout = sessionTimeout;
this.hostProvider = hostProvider;
this.chrootPath = chrootPath;
//保证一次会话超时时间内，可以重连全部的host
connectTimeout = sessionTimeout / hostProvider.size();
//读取超时为什么是会话超时的2/3
readTimeout = sessionTimeout * 2 / 3;
readOnly = canBeReadOnly;
//这里初始化了两个线程，一个用于发送，一个用于事件处理
//用于网络的通信的clientCnxnSocket类有两种，分别是原生NIO以及Netty
sendThread = new SendThread(clientCnxnSocket);
eventThread = new EventThread();
this.clientConfig=zooKeeper.getClientConfig();
}
//顺便一提的就是上面zk的构造函数中的start方法其实就是开启两个线程
public void start() {
sendThread.start();
eventThread.start();
}
看下通信线程sendThread，由于是线程，因此我们先从run方法开始，构造方法中有一个设置初始状态的逻辑：
//构造函数中的重点是修改了状态
SendThread(ClientCnxnSocket clientCnxnSocket) {
super(makeThreadName("-SendThread()"));
//初始化状态为连接中
state = States.CONNECTING;
this.clientCnxnSocket = clientCnxnSocket;
setDaemon(true);
}
run方法很长，一点一点看：
public void run() {
//初始化clientCnxnSocket中的变量，该类封装了底层的网络通信
clientCnxnSocket.introduce(this, sessionId, outgoingQueue);
//初始化当前时间
clientCnxnSocket.updateNow();
//更新最后一次发送数据包及最后一次心跳的时间，这里其实还没有发送过任何数据，为什么要更新时间？
clientCnxnSocket.updateLastSendAndHeard();
//从后面可以知道该变量表示未通信的时间间隔
int to;
//最后一次ping读写服务器的时间
long lastPingRwServer = Time.currentElapsedTime();
//发送ping数据包最大时间间隔
final int MAX_SEND_PING_INTERVAL = 10000; //10 seconds
InetSocketAddress serverAddress = null;
//构造方法中设置state为CONNECTING，所以这里是true
while (state.isAlive()) {
try {
//第一次的话这个条件判断肯定是true，后面如果连接断开则会在这里进行重连
if (!clientCnxnSocket.isConnected()) {
//此时如果已经关闭了，则不用再重新建立连接
if (closing) {
break;
}
//获取zk地址
//这里rwServerAddress表示读写服务器地址，跟zk的只读模式有关
if (rwServerAddress != null) {
serverAddress = rwServerAddress;
rwServerAddress = null;
} else {
//如果读写服务器地址为空，则从传入的地址列表中选一个
serverAddress = hostProvider.next(1000);
}
//连接服务器
startConnect(serverAddress);
//更新最后一次通信时间，同上
clientCnxnSocket.updateLastSendAndHeard();
}
//如果当前连接有效
if (state.isConnected()) {
//处理校验登录的场景，可以忽略
if (zooKeeperSaslClient != null) {
//省略掉了校验代码...
}
//这个分支说明连接已经建立，因此是剩余读取超时时间
to = readTimeout - clientCnxnSocket.getIdleRecv();
} else {
//这个分支说明连接还没有建立，因此是剩余连接超时时间
to = connectTimeout - clientCnxnSocket.getIdleRecv();
}
//看上面代码这个连接超时是根据会话超时时间算出来的，会话超时直接抛出异常，由后面cache住进行重连
if (to <= 0) {
throw new SessionTimeoutException(warnInfo);
}
//如果连接有效，则判断是否需要发送心跳信息
if (state.isConnected()) {
//计算是否需要发送ping包
int timeToNextPing = readTimeout / 2 - clientCnxnSocket.getIdleSend() -
((clientCnxnSocket.getIdleSend() > 1000) ? 1000 : 0);
if (timeToNextPing <= 0 || clientCnxnSocket.getIdleSend() > MAX_SEND_PING_INTERVAL) {
sendPing();
clientCnxnSocket.updateLastSend();
} else {
if (timeToNextPing < to) {
to = timeToNextPing;
}
}
}
//如果当前连接的是只读节点则寻找一个可写的服务器
if (state == States.CONNECTEDREADONLY) {
long now = Time.currentElapsedTime();
int idlePingRwServer = (int) (now - lastPingRwServer);
if (idlePingRwServer >= pingRwTimeout) {
lastPingRwServer = now;
idlePingRwServer = 0;
pingRwTimeout =
Math.min(2*pingRwTimeout, maxPingRwTimeout);
pingRwServer();
}
to = Math.min(to, pingRwTimeout - idlePingRwServer);
}
//开启网络通信
clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this);
} catch (Throwable e) {
//如果已经关闭则退出循环，即重试
if (closing) {
break;
} else {
//清理资源，包括关闭socket连接以及把队列中的请求任务按失败返回
cleanup();
//如果连接并没有关闭，则加入一个失去连接的事件
if (state.isAlive()) {
eventThread.queueEvent(new WatchedEvent(
Event.EventType.None,
Event.KeeperState.Disconnected,
null));
}
clientCnxnSocket.updateNow();
clientCnxnSocket.updateLastSendAndHeard();
}
}
}
//退出上面的循环表示连接断开，关闭IO资源以及如果等待队列中有数据包，则返回相应的失败事件
synchronized (state) {
//同上
cleanup();
}
clientCnxnSocket.close();
//如果之前连接是活的，则增加一个断开连接的事件
if (state.isAlive()) {
eventThread.queueEvent(new WatchedEvent(Event.EventType.None,
Event.KeeperState.Disconnected, null));
}
//增加一个连接关闭的事件
eventThread.queueEvent(new WatchedEvent(Event.EventType.None,
Event.KeeperState.Closed, null));
}
上面这个run方法首先要关注一下while循环，我可以看到在抛异常的情况下，除非已经显示的调用close()关闭连接，不然的话会做一部分清理工作然后继续尝试重新连接zk，也就是说只要不退出while循环就会一直重试连接zk，反过来说一旦退出循环则表示这次连接彻底结束，如果想重连ZK则需要重新初始化ZooKeeper对象。
通过看上面代码，除了显示调用close方法(该方法会把closing设置为true)，再就是state.isAlive()返回false，来看下这个方法：
public boolean isAlive() {
return this != CLOSED && this != AUTH_FAILED;
}
可以看到一个是AUTH_FAILED，也就是说如果校验失败则不会进行重新连接。
还有一个是CLOSED，这个在session过期时会设置，后面会看到代码，这里就记住会话过期也不会进行重新连接。
下面来看看IO交互部分的源码，IO部分首先要看下建立建立的代码，也就while循环中的startConnect方法：
private void startConnect(InetSocketAddress addr) throws IOException {
//如果不是第一次连接，即如果是重试连接则随机休息一下
if(!isFirstConnect){
try {
Thread.sleep(r.nextInt(1000));
} catch (InterruptedException e) {
LOG.warn("Unexpected exception", e);
}
}
//修改状态，这里之所以还要修改状态就是要考虑重试的情况
state = States.CONNECTING;
//这里省略校验连接跟日志的代码...
//开始正式请求连接，我们以NIO的代码为例
clientCnxnSocket.connect(addr);
}
void connect(InetSocketAddress addr) throws IOException {
//创建SocketChannel，以NIO为例，不多说了
SocketChannel sock = createSock();
try {
//连接并注册事件
registerAndConnect(sock, addr);
} catch (IOException e) {
LOG.error("Unable to open socket to " + addr);
sock.close();
throw e;
}
//初始化完成
initialized = false;
//lenBuffer是一个4字节的buffer，用于表示数据包的长度
//incomingBuffer是真正用于读取数据的buffer，把lenBuffer赋值给它表示
//读取数据时先读取4个字节(即lenBuffer的长度)，通过这4个字节来计算
//数据包的大小也就是后续还需要读取数据的长度，参考后面的IO交互逻辑
lenBuffer.clear();
incomingBuffer = lenBuffer;
}
//该方法最重要的逻辑就是注册了OP_CONNECT事件
void registerAndConnect(SocketChannel sock, InetSocketAddress addr) throws IOException {
sockKey = sock.register(selector, SelectionKey.OP_CONNECT);
boolean immediateConnect = sock.connect(addr);
if (immediateConnect) {
//如果连接马上建立，则调用该方法，否则OP_CONNECT事件也会调用，后面分析
sendThread.primeConnection();
}
}
通过上面建立连接的代码我们可以知道此时已经想服务端发送了建立连接的请求并注册了OP_CONNECT事件，下面开始看正式的IO交互逻辑，也就是while循环中的doTransport方法（这里是以NIO为例）：
void doTransport(int waitTimeOut, List<Packet> pendingQueue, ClientCnxn cnxn)throws IOException, InterruptedException {
//等待事件
selector.select(waitTimeOut);
Set<SelectionKey> selected;
synchronized (this) {
selected = selector.selectedKeys();
}
//更新当前时间
updateNow();
for (SelectionKey k : selected) {
SocketChannel sc = ((SocketChannel) k.channel());
//处理连接成功以后的事情，注意这里只是socket成功
if ((k.readyOps() & SelectionKey.OP_CONNECT) != 0) {
if (sc.finishConnect()) {
//更新数据包时间以及当前使用的地址
updateLastSendAndHeard();
updateSocketAddresses();
//发送连接请求给zk服务端，并且注册读写事件，后面看源码
sendThread.primeConnection();
}
} else if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
//可读或者可写，进行IO操作
doIO(pendingQueue, cnxn);
}
}
//如果有待发送的数据包，则注册写事件
if (sendThread.getZkState().isConnected()) {
if (findSendablePacket(outgoingQueue,
sendThread.tunnelAuthInProgress()) != null) {
enableWrite();
}
}
//清除事件，对NIO了解的话不难理解
selected.clear();
}
看看连接服务器成功以后做的事情，：
这里要特别注意的是调用该方法的是由于收到了CONNECT事件，但此时并没有注册读写事件，也就是说刚连到服务器，还没有开始数据包的发送
void primeConnection() throws IOException {
//一旦连接成功，则把首次连接变量置为false
isFirstConnect = false;
//由于新版的ZK是有只读模式的，但是只读模式下的sessionId是非法的
//这里的意思就是如果sessionId是从读写服务器获取的那就传给服务端
//否则就传0，关于只读模式的细节后面在分析服务端源码的时候再介绍
long sessId = (seenRwServerBefore) ? sessionId : 0;
//构造一个请求数据包，上面只是建立TCP建立，这里才是建立ZK应用层连接
ConnectRequest conReq = new ConnectRequest(0, lastZxid, sessionTimeout, sessId, sessionPasswd);
//连接成功以后，如果开启了watch自动重置则会把已有的watch重新在服务端注册一遍
if (!clientConfig.getBoolean(ZKClientConfig.DISABLE_AUTO_WATCH_RESET)) {
//这里如果是第一次连接，那么这几个集群都为空
//但如果是重新建立连接且开启了自动重新注册watcher则会把现有的watch重新发往服务端进行注册
List<String> dataWatches = zooKeeper.getDataWatches();
List<String> existWatches = zooKeeper.getExistWatches();
List<String> childWatches = zooKeeper.getChildWatches();
if (!dataWatches.isEmpty()
|| !existWatches.isEmpty() || !childWatches.isEmpty()) {
Iterator<String> dataWatchesIter = prependChroot(dataWatches).iterator();
Iterator<String> existWatchesIter = prependChroot(existWatches).iterator();
Iterator<String> childWatchesIter = prependChroot(childWatches).iterator();
long setWatchesLastZxid = lastZxid;
while (dataWatchesIter.hasNext()
|| existWatchesIter.hasNext() || childWatchesIter.hasNext()) {
List<String> dataWatchesBatch = new ArrayList<String>();
List<String> existWatchesBatch = new ArrayList<String>();
List<String> childWatchesBatch = new ArrayList<String>();
int batchLength = 0;
//分批重新注册watcher，避免大批量的watcher超过了包的大小限制
while (batchLength < SET_WATCHES_MAX_LENGTH) {
final String watch;
if (dataWatchesIter.hasNext()) {
watch = dataWatchesIter.next();
dataWatchesBatch.add(watch);
} else if (existWatchesIter.hasNext()) {
watch = existWatchesIter.next();
existWatchesBatch.add(watch);
} else if (childWatchesIter.hasNext()) {
watch = childWatchesIter.next();
childWatchesBatch.add(watch);
} else {
break;
}
batchLength += watch.length();
}
//为每批watch创建独立的数据包
SetWatches sw = new SetWatches(setWatchesLastZxid,
dataWatchesBatch,
existWatchesBatch,
childWatchesBatch);
RequestHeader header = new RequestHeader(-8, OpCode.setWatches);
Packet packet = new Packet(header, new ReplyHeader(), sw, null, null);
//加入发送队列开头，注意这里只是加入队列，由于还没有注册读写事件
//因此并没有开始发送
outgoingQueue.addFirst(packet);
}
}
}
//发验证信息加入发送队列开头
for (AuthData id : authInfo) {
outgoingQueue.addFirst(new Packet(new RequestHeader(-4,
OpCode.auth), null, new AuthPacket(0, id.scheme,
id.data), null, null));
}
//把连接请求数据包加入队列开头
outgoingQueue.addFirst(new Packet(null, null, conReq,
null, null, readOnly));
//里面是原生NIO代码，开始监听读写事件，这里才真正开始发送数据包
clientCnxnSocket.connectionPrimed();
}
上面这段数据包加入队列的逻辑由于都是调用的addFirst方法，因此顺序刚好是反过来的，最终结果如下图：
队列数据图
再看读写事件的IO交互，如果想顺着上面的代码看的话，这里建议先看对写事件的数据：
void doIO(List<Packet> pendingQueue, ClientCnxn cnxn)
throws InterruptedException, IOException {
SocketChannel sock = (SocketChannel) sockKey.channel();
if (sock == null) {
throw new IOException("Socket is null!");
}
//如果当前通道可读
if (sockKey.isReadable()) {
//先读取表示数据的长度前4个字节，此时incomingBuffer等于lenBuffer
int rc = sock.read(incomingBuffer);
if (rc < 0) {
throw new EndOfStreamException(
"Unable to read additional data from server sessionid 0x"
+ Long.toHexString(sessionId)
+ ", likely server has closed socket");
}
//如果成功读取到了前4个字节，注意incomingBuffer变量初始化就是4个字节大小
if (!incomingBuffer.hasRemaining()) {
incomingBuffer.flip();
//这个条件判断表示当前读的值是否为表示数据长度的4个字节
if (incomingBuffer == lenBuffer) {
recvCount++;
//根据实际数据长度初始化数组，该方法就是修改incomingBuffer的长度为实际数据的长度
readLength();
} else if (!initialized) {
//如果尚未进行初始化则进入这个分支
//初始化服务端返回的信息，例如sessionId，会话超时时间等
readConnectResult();
//监听读事件
enableRead();
//如果有必要则激活写事件，这里是因为首次建立建立的话，即使有数据包也不能发送，要等到收到
//服务端返回的响应才开始发送
if (findSendablePacket(outgoingQueue, sendThread.tunnelAuthInProgress()) != null) {
enableWrite();
}
//清空lenBuffer，incomingBuffer两个数据，准备读取下一个数据包
lenBuffer.clear();
incomingBuffer = lenBuffer;
updateLastHeard();
initialized = true;
} else {
//读取数据，正式解析数据包中内容
sendThread.readResponse(incomingBuffer);
//清空lenBuffer，incomingBuffer两个数据，准备读取下一个数据包
lenBuffer.clear();
incomingBuffer = lenBuffer;
updateLastHeard();
}
}
}
//如果当前通道可写
if (sockKey.isWritable()) {
//寻找可发送的包
Packet p = findSendablePacket(outgoingQueue,
sendThread.tunnelAuthInProgress());
if (p != null) {
//更新最后发送时间
updateLastSend();
if (p.bb == null) {
if ((p.requestHeader != null) &&
(p.requestHeader.getType() != OpCode.ping) &&
(p.requestHeader.getType() != OpCode.auth)) {
////初始化xid，这里可以看到连接，ping，auth三种数据包是不占用xid的
p.requestHeader.setXid(cnxn.getXid());
}
p.createBB();
}
sock.write(p.bb);
//数据包发送完成则计数并加入队列
if (!p.bb.hasRemaining()) {
sentCount++;
outgoingQueue.removeFirstOccurrence(p);
if (p.requestHeader != null
&& p.requestHeader.getType() != OpCode.ping
&& p.requestHeader.getType() != OpCode.auth) {
synchronized (pendingQueue) {
//加入等待响应的队列，这里可以看到连接，ping，auth三种数据包是不会强制顺序接收响应的
pendingQueue.add(p);
}
}
}
}
//如果没有可发送数据，则取消监听写事件
if (outgoingQueue.isEmpty()) {
disableWrite();
} else if (!initialized && p != null && !p.bb.hasRemaining()) {
//初始化还没完成且连接的请求已经发送，则阻止后面的请求继续发送
//也就是说只有应用层连接建立完成以后才开始后续的数据包发送
disableWrite();
} else {
//其他情况则激活写事件
enableWrite();
}
}
}
上面的代码片段中readConnectResult()方法就是发送完建立请求的连接以后，读取服务端返回的sessionId，超时时间，当前模式（只读还是读写）等信息，逻辑比较简单就不贴代码了，我们重点来看下读取普通数据的逻辑，即readResponse方法，上代码：
void readResponse(ByteBuffer incomingBuffer) throws IOException {
ByteBufferInputStream bbis = new ByteBufferInputStream(incomingBuffer);
BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
ReplyHeader replyHdr = new ReplyHeader();
//反序列化头信息
replyHdr.deserialize(bbia, "header");
//-2表示服务端响应的是客户端的ping包
if (replyHdr.getXid() == -2) {
return;
}
//-4表示服务端响应客户端校验连接的请求
if (replyHdr.getXid() == -4) {
if(replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) {
state = States.AUTH_FAILED;
//添加校验失败的事件
eventThread.queueEvent( new WatchedEvent(Watcher.Event.EventType.None,
Watcher.Event.KeeperState.AuthFailed, null) );
//终止事件
eventThread.queueEventOfDeath();
}
return;
}
//-1表示通知响应，比如watch事件，数据被修改以后服务端主动通知客户端
if (replyHdr.getXid() == -1) {
WatcherEvent event = new WatcherEvent();
//反序列化事件信息
event.deserialize(bbia, "response");
//如果传入的zk地址指定了初始路径，则要做转换
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);
}
}
//封装watch事件并加入队列
WatchedEvent we = new WatchedEvent(event);
eventThread.queueEvent( we );
return;
}
//如果正在校验登录信息则直接给服务端响应，可忽略
if (tunnelAuthInProgress()) {
GetSASLRequest request = new GetSASLRequest();
request.deserialize(bbia,"token");
zooKeeperSaslClient.respondToServer(request.getToken(),ClientCnxn.this);
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");
}
} finally {
//把数据包的收尾工作放在这里处理
finishPacket(packet);
}
}
为了更好的理解watch事件的处理以及容易理解后面的eventThread逻辑，这里先看下上面包装watch事件的方法queueEvent，如下
public void queueEvent(WatchedEvent event) {
queueEvent(event, null);
}
private void queueEvent(WatchedEvent event,
Set<Watcher> materializedWatchers) {
if (event.getType() == EventType.None
&& sessionState == event.getState()) {
return;
}
sessionState = event.getState();
final Set<Watcher> watchers;
if (materializedWatchers == null) {
//这里就是获取所有关心该事件的watcher，而这个watcher就是在初始化的时候复制的watchManager
//默认实现是ZKWatchManager，后面再分析
watchers = watcher.materialize(event.getState(),
event.getType(), event.getPath());
} else {
watchers = new HashSet<Watcher>();
watchers.addAll(materializedWatchers);
}
//这里把多个watcher封装成WatcherSetEventPair对象，
//重点就是这里真正加入队列的watch类型是WatcherSetEventPair
WatcherSetEventPair pair = new WatcherSetEventPair(watchers, event);
waitingEvents.add(pair);
}
好了，看完对watch事件的包装逻辑，我们来看下上面IO代码片段中的最后一个方法finishPacket，如下：
private void finishPacket(Packet p) {
int err = p.replyHeader.getErr();
if (p.watchRegistration != null) {
//如果服务端处理watcher成功，则在客户端注册，此时watcher才真正生效
//换句话说，用户注册watcher时，客户端会先把请求发在服务端，只有
//服务端注册成功，该watcher对象在客户端才会生效
p.watchRegistration.register(err);
}
//如果有取消注册的watcher，则封装成事件通知客户端
if (p.watchDeregistration != null) {
Map<EventType, Set<Watcher>> materializedWatchers = null;
try {
materializedWatchers = p.watchDeregistration.unregister(err);
for (Entry<EventType, Set<Watcher>> entry : materializedWatchers
.entrySet()) {
Set<Watcher> watchers = entry.getValue();
if (watchers.size() > 0) {
queueEvent(p.watchDeregistration.getClientPath(), err,
watchers, entry.getKey());
p.replyHeader.setErr(Code.OK.intValue());
}
}
} catch (KeeperException.NoWatcherException nwe) {
p.replyHeader.setErr(nwe.code().intValue());
} catch (KeeperException ke) {
p.replyHeader.setErr(ke.code().intValue());
}
}
//如果是同步处理，则唤醒等待线程
if (p.cb == null) {
synchronized (p) {
p.finished = true;
p.notifyAll();
}
} else {
//异步处理数据包
p.finished = true;
eventThread.queuePacket(p);
}
}
好了，到目前为止，整个zk客户端建立连接，数据包交互的逻辑就都看完了，接下来需要看另一个关键点，事件处理的逻辑，上代码eventThread的run方法：
public void run() {
try {
isRunning = true;
while (true) {
Object event = waitingEvents.take();
//如果收到的事件表示当前会话结束，则停止时间的处理
if (event == eventOfDeath) {
wasKilled = true;
} else {
//处理事件
processEvent(event);
}
if (wasKilled)
synchronized (waitingEvents) {
if (waitingEvents.isEmpty()) {
isRunning = false;
break;
}
}
}
} catch (InterruptedException e) {
LOG.error("Event thread exiting due to interruption", e);
}
}
上面的processEvent方法就是按照事件类型处理，包括watcher通知很直观，这里就不贴代码了。
最后我们来看下watcher的管理逻辑：
public interface ClientWatchManager {
//从上面的代码可以看出来，在收到服务端的响应时，客户端会调用该方法
//来获取所有对该事件感兴趣的watcher
public Set<Watcher> materialize(Watcher.Event.KeeperState state,
Watcher.Event.EventType type, String path);
}
代码很简单，其实就是根据服务端返回的数据包，把事件类型以及路径传入，然后由客户端自己返回关心这次事件的watcher由于后面触发，这里可以看出我们注册watcher是客户端自己管理的。ZooKeeper中有个默认的实现，代码比较长，这里就不贴的，总的来说就是它把watcher分为dataWatches，existWatches跟childWatches三类并按照路径组织，这里再根据服务端返回的事件对应的path取出相应的watcher然后触发。
上面没有贴出实际使用zk的API的代码，比如读取某个节点的值，这是因为当你搞清楚上面说的整个通信逻辑，事件处理逻辑以后再看对用户开放的那些API代码就会觉得非常简单，感兴趣的话可以去看看，这里就不多说了。
三、总结
以上就是zk客户端的关键流程，如果有些细节的地方觉得不好理解，这很正常，因为你还不了解服务端的处理流程，很快我会再写一篇解析zk服务端源码的文章，相信到时候一定能够更好的理解zk的原理。
又是一篇很长的文章，不知道有几个人可以耐心的看完~
