9. 源码| 交互流程分析
大约 24 分钟
对于ZK在运行时有诸多疑问，ZK对于IO多路复用是如何使用的？Client端和Server端是通过什么方式实现了交互协议的？以及在ZK的运行过程中起到关键作用的组件和不同操作的交互流程又是怎样的？
本篇文章将会从Client端、Server端的交互流程入手，先分析两端的交互流程以及重要组件，随后再仔细分析这些重要组件的作用及生效方式来解决上述的几个问题。需要注意的是交互操作分析的是新建连接，对于新建节点、ping以及关闭连接等的操作大致流程和新建是差不多的，只是内部的操作稍微有点变化而已，这些后续再来分析。
注：本篇基于ZK版本3.7分析的，且需要对ZK的架构以及重要组件组成有一定的基础了解，当然先看完流程再去了解那些架构组件也行，但还是推荐先去了解Server端以及Client端的架构组件，因为本篇不会进行一些基础性的分析。
1. 交互流程
对于ZK来说，Client和Server端的交互流程值得学习，无论是对于NIO或者Netty的使用，还是为了解决通信数据传递问题。
本次交互流程只考虑正常连接情况，并且将其流程拆解为三步走，分别为：1、客户端发起连接Server端请求；2、Server端收到并处理响应Client端的连接请求；3、收到Server端的响应生成对应的响应事件触发本地的监听器。大致交互流程图如下：
接下来看看三步走的具体详细交互流程图。
1.1 Client端发起连接
三步走中的第一步具体详细流程图如下：
在这个步骤中担任主要职责的便是SendThread，图中简单的把流程分为了11个步骤，这个图中如果看过了上两篇关于ZK服务和客户端端的重要组件便可以清晰都知道其大致作用，只是图中加入了NIO的一些类而已。接下来具体分析一下在各个步骤中的一些小细节：
- C1：对应着Java应用程序传入链接串并实例化ZooKeeper类进行连接；
- C2：说明ZooKeeper只是一个API类而已，实际和Server端的交互逻辑并不在这个类中，这个类只会封装或设置请求参数并交给ClientCnxn实际连接对象处理；
- C3：这个步骤总的来说是初始化并启用SendThread和EventThread两个线程对象，SendThread线程负责轮询NIO事件和心跳检测，EventThread负责处理Server端响应的事件，在图中分为了四个步骤是为了更直观的看具体的操作以及大致作用；
- C4：实际上有非常多的逻辑：更新并判断处理心跳检测时间、在刚开始的时候尝试第一次连接Server端服务、以及判断验证信息等。并且SendThread是一个线程对象，C4这个流程将会被反复调用，这也就是为什么途中有C4.1和C4.2了，C4.1对应C5、C6、和C7，而C4.2则对应C8那条链路的流程；
- C5：为什么这里要说尝试第一次连接？因为在这个流程中会调用一次SocketChannel的连接方法connect，而SocketChannel在创建的时候就已经被设置为阻塞状态了，因此connect方法即使没有立即连上去，也会后续通过一个OP_CONNECT事件来通知，但也有几率第一次就连上去，虽然几率比较小，下面的C6和C7便是针对C5第一次没连接上而采取的措施；
- C6：当第一次尝试连接没连上，后续NIO的Server端连上了Client端的Selector将会收到NIO事件OP_CONNECT，并获取到相应的SelectionKey和SocketChannel对象；
- C7：获取到NIO的Socket后将会完成连接，并且生成对应的ConnectRequest和Packet对象放到outgoingQueue数组中以便下次开启OP_WRITE写事件后能够发送数据包；
- C8：这个流程应该要从C3.3开始走起，当C7完成连接并且把数据包保存在outgoingQueue数组中时，将会开启OP_WRITE事件，此时SendThread循环时将可以通过Selector获取到这个写事件，从而进入C8流程判断为写操作；
- C9：当确认进入写操作时将会把Packet取出来以便后续流程使用；
- C10：调用C9流程拿出来的Packet对象的createBB()方法，将其序列化并存放到Packet对象中的ByteBuffer缓存对象中；
- C11：使用获取到的SocketChannel对象将ByteBuffer缓存对象中的序列化数据发送至Server端，并随后继续判断outgoingQueue和Packet对象是否还有数据，如果有数据则保持OP_WRITE事件开启，否则关闭OP_WRITE，只进行监听Server端的数据。 如果对NIO的交互流程有一定的了解，对于ZK为何要这样实现的应该能理解一二，如果对NIO交互流程不怎么熟悉的，也可以参照ZK的使用，自己写一个通信多路复用的Demo。
1.2 Server接收处理及响应
三步走中的第二步Server端交互处理流程如下：
从ZK系列的第二篇文章可以知道NIOServerCnxnFactory在ZK启动时也是以一个守护线程对象运行的，会一直通过Selector轮询是否有新的IO事件，如果有则根据IO事件类型进行相应的处理，接下来详细分析下其具体的交互流程：
- S1：守护线程对象将会每隔1s使用Selector轮询是否有新的IO事件；
- S2：当Client端调用了SocketChannel.connect()方法时，Selector将会收到OP_ACCEPT连接类型的NIO事件，并获取对应的SocketChannel和SelectionKey对象；
- S3：当确认是OP_ACCEPT事件时，将会先判断是否到达最大连接数了，满足则不会创建新的连接对象，否则注册SocketChannel生成SelectionKey，并设置成OP_READ读模式，根据这两个对象创建NIO连接对象NIOServerCnxn；
- S4：会将NIOServerCnxn和SelectionKey进行绑定，并将NIOServerCnxn添加到cnxns数组和ip-NIOServerCnxn对应关系的map对象ipMap，执行完该流程将会监听等待Client端后续的请求；
- S5：执行到这个流程时说明Client端已经执行到了C11（即发送具体的连接请求）流程了，此时Server端将会收到来自Client端Socket的IO事件；
- S6：前面收到的IO事件对应的SelectionKey操作类型是OP_READ，将会获取和其绑定的NIOServerCnxn对象；
- S7：这一步会调用NIOServerCnxn对象的doIO()方法，这里面将会根据是否初始化来判断是读取连接请求还是普通的请求，当然在我们这个流程中读取的是连接请求；
- S8：将接收到的ByteBuffer反序列化成ConnectRequest请求对象，并根据Server端的心跳间隔时间以及Client传过来的SessionTimeout过期时间做一个中和判断，得出session的过期时间，并利用前面获得的NIOServerCnxn以及sesion过期时间在SessionTracker中创建session并进行跟踪，后续在分析ping心跳检测操作时再详细分析；
- S9：根据已有对象信息创建Request对象，这个对象代表Client的每次具体请求，请求的内容以及相关的session信息都会在这个类中，并且后续的RequestProcessor系列对象处理最小单元便是Request对象类型，调用下一个流程前会更新一波session的过期时间；
- S10：这个流程的具体执行是在RequestProcessor处理器实现类PrepRequestProcessor中完成的，其也是一系列实现类中的第一个执行类，，主要完成的操作便是根据Request中的header对象操作类型type属性来创建对应的CreateSessionTxn对象；
- S11：执行完S10后S11做的操作只有更新一下session失效时间，其它的流程在单机运行中并未起到很大作用；
- S12：此时已经调用到了第二个RequestProcessor处理器SyncRequestProcessor中，这个处理器做的事情便是保存请求日志和运行快照，具体的处理细节后续看有机会再仔细分析一波；
- S13：当完成对logDir位置进行日志新增时，将会调用到下一个RequestProcessor处理器- FinalRequestProcessor中，在这里面完成最后的处理及响应；
- S14：这个流程是同步的，并且在这个流程中也会刷新一次session过期时间，并刷新ZK的serverStatus和NIOServerCnxn的近期调用状态时间等；
- S15：最后把session过期时间、sessionId以及生成的随机密码等序列化到ByteBuffer中，随后通过SocketChannel写入到IO通道中通知Client端。这是正常流程，还有一种便是响应太长导致一次性发不完便会再次使用NIO的Selector.select()方法处理自身产生的写事件，直到把响应全部写完。直到这个流程Server端的连接处理响应流程便全部走完。
在跟踪这一次请求源码时ZK进行了多次刷新session过期时间，为什么ZK要在各个Request处理器中都进行一次刷新session过期时间呢？以并发量小的角度看这个问题可能会很不解，因为针对仅有一次的请求情况，各个处理器之间相当于是同步处理的，所以看起来没有那么大的必要；但如果ZK的并发量高了起来，单机部署的情况下除了SyncRequestProcessor调用FinalRequestProcessor是同步流程，其它的都是线程异步的，一个响应由ZooKeeperServer类调用到FinalRequestProcessor可能中间会相差比较长的时间，如果只在开始调用或者结束调用的地方进行session过期时间刷新，中途可能session追踪器便已经把那些过期时间短的session当做过期的处理了。
当然上述只是我的猜测，ZK获取有更多的考量，其它方面的原因便需要后续对ZK的深入了解才能知道了。
1.3 Client端接收Server端响应
三步走中的最后一步交互流程图如下：
这个流程相对于前两步而言步骤不是很多，大致就两点：1、Client端监听收到响应；2、触发本地的监听器对发生的ZK事件进行处理。具体流程分析如下：
- C1：这个步骤可以看成两个同时进行的步骤：C1.1为SendThread线程轮询Selector监听Server端是否有新的响应，C2.2为EventThread监听waitingEvents数组是否有等待事件处理，需要注意的是waitingEvents数组中的元素只会通过SendThread线程收到响应处理后添加进去，因此waitingEvents数组的来源可以看成就是SendThread添加的容易理解一点；
- C2：只针对新建连接而言，这个步骤获取到的IO事件为OP_READ；
- C3：判断IO事件的类型，将会进入doIO()方法读取SocketChannel的数据；
- C4：使用前面读取到的ByteBuffer数据，反序列化成ConnectResponse对象；
- C5：根据响应对象的属性设置心跳检测需要的属性，如readTimeout、connectTimeout和negotiatedSessionTimeout等，最后会根据KeeperState生成WatchedEvent对象；
- C6：通过WatchedEvent的KeeperState更新session的状态；
- C7：根据ClientWatchManager以及传入进来的WatchedEvent生成WatcherSetEventPair对象，保存了需要触发的监听器以及对应的响应事件；
- C8：将WatcherSetEventPair添加到waitingEvents数组中，waitingEvents数组中的对象也有可能是Packet类型的对象；
- C9：通过C1.2开始一直轮询waitingEvents数组，当完成C8之后，C9将可以轮询到事件对象WatcherSetEventPair并进行处理；
- C10：如果event对象不是eventOfDeath（关闭EventThread线程对象标识）对象，则会判断是否为WatcherSetEventPair类型对象，如果是则遍历对象中的watcher对象，并传入事件对象WatchedEvent进行回调；否则会调用Packet中的AsyncCallback对象进行异步回调。
截止到这里回调基本上就已经结束了，第三步无论是新建连接还是触发事件进行回调流程都是一样的，都是一直轮询waitingEvents数组，并判断类型调用相应的监听器。后续分析操作命令交互以及ping等操作时这一步都是通用的。
2. Client发起连接源码分析
2.1 ZooKeeper入口类
前面说过，ZooKeeper是ZK客户端的API类，连接以及其它的操作都是以这个类为入口的，接下来看下其新建连接的对外接口：
public class ZooKeeper {
protected final ClientCnxn cnxn;
private final ZKWatchManager watchManager = new ZKWatchManager();
public ZooKeeper(String connectString, int sessionTimeout,
Watcher watcher) throws IOException {
// 一般而言新建连接都是使用的这个接口
this(connectString, sessionTimeout, watcher, false);
}
public ZooKeeper(
String connectString,
int sessionTimeout,
Watcher watcher,
boolean canBeReadOnly) throws IOException {
// createDefaultHostProvider() 解析给出的server的址址，并对解析结果进行第一次shuffle
this(connectString, sessionTimeout, watcher, canBeReadOnly, createDefaultHostProvider(connectString));
}
private static HostProvider createDefaultHostProvider(String connectString) {
// ConnectStringParser() 用于解析指定的server地址列表字符串
// getServerAddresses() 获取到所有解析出来的地址集合
return new StaticHostProvider(new ConnectStringParser(connectString).getServerAddresses());
}
public ZooKeeper(
String connectString,
int sessionTimeout,
Watcher watcher,
boolean canBeReadOnly,
HostProvider hostProvider,
ZKClientConfig clientConfig
) throws IOException {
LOG.info(
"Initiating client connection, connectString={} sessionTimeout={} watcher={}",
connectString,
sessionTimeout,
watcher);
validateWatcher(watcher);
this.clientConfig = clientConfig != null ? clientConfig : new ZKClientConfig();
this.hostProvider = hostProvider;
// 创建一个zk集群字符串解析器，将解析出的ip与port构建为一个地址实例，放入到缓存集合
ConnectStringParser connectStringParser = new ConnectStringParser(connectString);
// 创建一个对server的连接
cnxn = createConnection(
connectStringParser.getChrootPath(),
hostProvider,
sessionTimeout,
this.clientConfig,
watcher,
getClientCnxnSocket(),
canBeReadOnly);
// 开始连接
cnxn.start();
}
}createDefaultHostProvider(connectString)创建主机提供者，把将缓存集合中的地址打散:
public StaticHostProvider(Collection<InetSocketAddress> serverAddresses) {
// init() 的第三个参数是创建了一个地址处理器
// init()中进行了第一次地址的shuffle
init(serverAddresses, System.currentTimeMillis() ^ this.hashCode(), new Resolver() {
// 根据指定的主机名，获取到所有其对应的ip地址
@Override
public InetAddress[] getAllByName(String name) throws UnknownHostException {
return InetAddress.getAllByName(name);
}
});
private void init(Collection<InetSocketAddress> serverAddresses, long randomnessSeed, Resolver resolver) {
this.sourceOfRandomness = new Random(randomnessSeed);
this.resolver = resolver;
if (serverAddresses.isEmpty()) {
throw new IllegalArgumentException("A HostProvider may not be empty!");
}
// 对地址的第一次打散(shuffle)
this.serverAddresses = shuffle(serverAddresses);
currentIndex = -1;
lastIndex = -1;
}
}
打散的目的在于负载均衡，不然每个客户端轮询都会连上第一个
2.2 ClientCnxn连接交互类
这个类里面有EventThread和SendThread，这两个内部类是ZK交互时最重要的两个类，前面也提过，接下来看下ClientCnxn是如何启动初始化这两个内部线程类的。
public class ClientCnxn {
// 当Client端的数据包Packet被发送出去时，如果不是ping和auth两种操作类型，其
// 它操作类型的包都会保存在队列末尾，代表着已发送但未完成的数据，在最后Client
// 端收到ZK的响应时，将会把队列第一个拿出来进行响应的处理。采用的是FIFO模式，
// 是因为ZK的Server端接收请求处理请求是有序的，处理完前面一个才会处理后面一个
// 因此客户端可以采用FIFO的模式处理
private final LinkedList<Packet> pendingQueue =
new LinkedList<Packet>();
// 发送队列，当Client端有请求需要发送时将会封装成Packet包添加到这里面，在
// SendThread线程轮询到有数据时将会取出第一个包数据进行处理发送。使用的也是
// FIFO模式
private final LinkedList<Packet> outgoingQueue =
new LinkedList<Packet>();
// 连接时间，初始化时等于客户端sessionTimeout / 可用连接串数量，如果连接成功
// 后将会等于协约时间negotiatedSessionTimeout / 可用连接串数量，因此正常
// 而言，此值就是negotiatedSessionTimeout / 可用连接串数量
private int connectTimeout;
// 协约时间，ZK的Server端会设置tickTime，Client端会传sessionTimeout，ZK的
// Server端将会根据两边的配置进行计算得出两边都能接受的时间，然后返回。这个
// 字段保存的就是协商之后的session过期时间
private volatile int negotiatedSessionTimeout;
// 读取过期时间，连接时值为sessionTimeout * 2 / 3，当连接成功后值为
// negotiatedSessionTimeout * 2 / 3
private int readTimeout;
// 开发人员自己定义的客户端过期时间sessionTimeout（注意这个时间并不是最终
// Client端运行时的心跳检测时间，后续会出一篇这些时间的具体作用以及计算规则）
private final int sessionTimeout;
// 入口类的引用对象
private final ZooKeeper zooKeeper;
// 客户端的监听器管理类，包含了默认监听器和三种不同类型的监听器
private final ClientWatchManager watcher;
// 本客户端连接实例的sessionId
private long sessionId;
// 是否只可读
private boolean readOnly;
// 将来将会被删除，暂时不知道有何用
final String chrootPath;
// Client端对Server端发送和接收消息的线程对象
final SendThread sendThread;
// Client端负责处理响应事件的线程对象
final EventThread eventThread;
// Client端的连接是否已经关闭
private volatile boolean closing = false;
// 连接串的解析后获得的InetSocketAddress提供对象
private final HostProvider hostProvider;
public ClientCnxn(String chrootPath, HostProvider hostProvider,
int sessionTimeout, ZooKeeper zooKeeper,
ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,
boolean canBeReadOnly) throws IOException {
// 没有连接密码的构造函数
this(chrootPath, hostProvider, sessionTimeout, zooKeeper, watcher,
clientCnxnSocket, 0, new byte[16], canBeReadOnly);
}
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;
// 计算未连接时的过期时间
connectTimeout = sessionTimeout / hostProvider.size();
readTimeout = sessionTimeout * 2 / 3;
readOnly = canBeReadOnly;
// 初始化两个线程对象
sendThread = new SendThread(clientCnxnSocket);
eventThread = new EventThread();
}
public void start() {
// 分别启动两个内部线程类
sendThread.start();
eventThread.start();
}
}2.3 SendThread发送连接请求
在ClientCnxn中启动SendThread线程后接下来的主角便只是SendThread以及调用的类了，而EventThread类只是在处理事件对象时会分析到。这个类是通过一直循环来进行不同的操作，因此不要把这个流程看成只有单一的功能，接收、发送以及ping等操作都是在循环中完成的，但现在我们只分析发送连接请求的代码。
class SendThread extends ZooKeeperThread {
// 客户端连接Server端的负责对象，默认采用的是NIO方式连接
private final ClientCnxnSocket clientCnxnSocket;
// 是否为第一次连接，默认是true
private boolean isFirstConnect = true;
@Override
@SuppressFBWarnings("JLM_JSR166_UTILCONCURRENT_MONITORENTER")
public void run() {
// 更新clientCnxnSocket的发送事件以及关联SendTreahd，这里sessionId
// 没有值，就是0
clientCnxnSocket.introduce(this, sessionId, outgoingQueue);
clientCnxnSocket.updateNow();
clientCnxnSocket.updateLastSendAndHeard();
// 上次ping和现在的时间差
int to;
long lastPingRwServer = Time.currentElapsedTime();
final int MAX_SEND_PING_INTERVAL = 10000; //10 seconds
InetSocketAddress serverAddress = null;
// 只要连接没有关闭，也没有验证失败，就一直循环
while (state.isAlive()) {
try {
// 刚开始运行时这里肯定是未连接的状态，因此会进去
if (!clientCnxnSocket.isConnected()) {
// don't re-establish connection if we are closing
// 如果ZK已经关闭了则直接会出循环
if (closing) {
break;
}
if (rwServerAddress != null) {
serverAddress = rwServerAddress;
rwServerAddress = null;
} else {
// 获取要连接的server的址
serverAddress = hostProvider.next(1000);
}
onConnecting(serverAddress);
// 开始连接
startConnect(serverAddress);
// Update now to start the connection timer right after we make a connection attempt
clientCnxnSocket.updateNow();
// 更新交互（连接请求/读写请求）时间戳
clientCnxnSocket.updateLastSendAndHeard();
}
...
// 获取已经有多久没有收到交互响应了
to = readTimeout - clientCnxnSocket.getIdleRecv();
} else {
// 获取已经有多久没有收到连接请求的响应了
to = connectTimeout - clientCnxnSocket.getIdleRecv();
}
// 处理会话超时的情况
if (to <= 0) {
String warnInfo = String.format(
"Client session timed out, have not heard from server in %dms for session id 0x%s",
clientCnxnSocket.getIdleRecv(),
Long.toHexString(sessionId));
LOG.warn(warnInfo);
// 抛出会话超时异常
throw new SessionTimeoutException(warnInfo);
}
...
// 这个方法十分重要，因为不管是连接还是其它任何操作都会进入
// 该方法进行操作类型判断已经发送接收数据包，具体流程留到
// 后续分析clientCnxnSocket对象时再看
clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this);
} catch (Throwable e) {
... 异常处理
}
}
// 跑到这里说明ZK已经关闭了，后面会做一些善后的工作，如发送关闭事件
// 清除连接的缓存数据等
synchronized (outgoingQueue) {
// When it comes to this point, it guarantees that later queued
// packet to outgoingQueue will be notified of death.
cleanup();
}
clientCnxnSocket.close();
...
}
private void startConnect(InetSocketAddress addr) throws IOException {
...
// 修改server状态
changeZkState(States.CONNECTING);
String hostPort = addr.getHostString() + ":" + addr.getPort();
MDC.put("myid", hostPort);
// 设置连接名称
setName(getName().replaceAll("\(.*\)", "(" + hostPort + ")"));
// 判断是否开启了SASL的客户端验证机制（C/S模式的验证机制）
if (clientConfig.isSaslClientEnabled()) {
...
}
// 进行连接的日志打印
logStartConnect(addr);
// 调用clientCnxnSocket的连接方法
clientCnxnSocket.connect(addr);
}
void primeConnection() throws IOException {
// 调用了这个方法说明客户端和Server端的Socket长连接已经连接完毕了
// 设置isFirstConnect为false
isFirstConnect = false;
long sessId = (seenRwServerBefore) ? sessionId : 0;
// 创建连接的请求对象ConnectRequest
ConnectRequest conReq = new ConnectRequest(0, lastZxid, sessionTimeout, sessId, sessionPasswd);
// We add backwards since we are pushing into the front
// Only send if there's a pending watch
// disableAutoWatchReset对应着ZK的启动属性
// zookeeper.disableAutoWatchReset，如果为false则为自动将ZK的
// 监听器监听到相应的节点，为true则不会自动监听
if (!clientConfig.getBoolean(ZKClientConfig.DISABLE_AUTO_WATCH_RESET)) {
// 接下来的流程大概就是从zooKeeper获取三种类型的监听器
// 把三种类型的监听器依次封装成SetWatches包保存到
// outgoingQueue包中以便后续发送包数据，具体的流程便忽略
List<String> dataWatches = watchManager.getDataWatchList();
List<String> existWatches = watchManager.getExistWatchList();
List<String> childWatches = watchManager.getChildWatchList();
List<String> persistentWatches = watchManager.getPersistentWatchList();
List<String> persistentRecursiveWatches = watchManager.getPersistentRecursiveWatchList();
if (!dataWatches.isEmpty() || !existWatches.isEmpty() || !childWatches.isEmpty()
|| !persistentWatches.isEmpty() || !persistentRecursiveWatches.isEmpty()) {
...
// 轮询三种的迭代器获取迭代器具体数据
while (dataWatchesIter.hasNext() || existWatchesIter.hasNext() || childWatchesIter.hasNext()
|| persistentWatchesIter.hasNext() || persistentRecursiveWatchesIter.hasNext()) {
...
// 将获取到的监听器封装成SetWatches对象
Record record;
int opcode;
if (persistentWatchesBatch.isEmpty() && persistentRecursiveWatchesBatch.isEmpty()) {
// maintain compatibility with older servers - if no persistent/recursive watchers
// are used, use the old version of SetWatches
record = new SetWatches(setWatchesLastZxid, dataWatchesBatch, existWatchesBatch, childWatchesBatch);
opcode = OpCode.setWatches;
} else {
record = new SetWatches2(setWatchesLastZxid, dataWatchesBatch, existWatchesBatch,
childWatchesBatch, persistentWatchesBatch, persistentRecursiveWatchesBatch);
opcode = OpCode.setWatches2;
}
RequestHeader header = new RequestHeader(ClientCnxn.SET_WATCHES_XID, opcode);
// 随后使用Packet封装Header和Recrod
Packet packet = new Packet(header, new ReplyHeader(), record, null, null);
// 添加到outgoingQueue数据中
outgoingQueue.addFirst(packet);
}
}
}
...
// 将ConnectRequest同样封装成Packet对象放到outgoingQueue中
outgoingQueue.addFirst(new Packet(null, null, conReq, null, null, readOnly));
// 开启OP_WRITE操作，开启后Selector.select()将可以收到读IO
clientCnxnSocket.connectionPrimed();
LOG.debug("Session establishment request sent on {}", clientCnxnSocket.getRemoteSocketAddress());
}
}从源码可以看出来SendThread只是一个线程轮询调用类，具体的发送和接收操作是交给ClientCnxnSocket对象来完成的。
2.4 ClientCnxnSocket套接字交互类
和Socket进行交互的类，负责向Socket中写入数据和读取数据。在连接流程中最重要的两个方法connect和doTransport都是在这个类中，根据在SendThread类中的流程，我们先分析connect，再去看doTransport方法。
public class ClientCnxnSocketNIO extends ClientCnxnSocket {
// NIO的多路复用选择器
private final Selector selector = Selector.open();
// 本Socket对应的SelectionKey
private SelectionKey sockKey;
@Override
void connect(InetSocketAddress addr) throws IOException {
// 创建一个NIO的channel
SocketChannel sock = createSock();
try {
// 这个方法的作用便是注册并尝试进行连接
registerAndConnect(sock, addr);
} catch (IOException e) {
// 注册socket失败
...
}
// 设置为非初始化
initialized = false;
lenBuffer.clear();
incomingBuffer = lenBuffer;
}
SocketChannel createSock() throws IOException {
// 创建一个SocketChannel对象，并设置非阻塞以及其它属性
SocketChannel sock;
sock = SocketChannel.open();
sock.configureBlocking(false);
sock.socket().setSoLinger(false, -1);
sock.socket().setTcpNoDelay(true);
return sock;
}
void registerAndConnect(SocketChannel sock, InetSocketAddress addr)
throws IOException {
// 将Socket注册到Selector中，并生成唯一对应的SelectionKey对象
sockKey = sock.register(selector, SelectionKey.OP_CONNECT);
// 进行Socket连接
boolean immediateConnect = sock.connect(addr);
// 如果第一次调用就已经连接上，则执行主要的连接操作
if (immediateConnect) {
// 这个方法前面已经介绍过了
sendThread.primeConnection();
}
}
@Override
void doTransport(int waitTimeOut, List<Packet> pendingQueue,
LinkedList<Packet> outgoingQueue, ClientCnxn cnxn)
throws IOException, InterruptedException {
// 最多休眠waitTimeOut时间获取NIO事件，调用wake()方法、有可读IO事件和
// 有OP_WRITE写事件可触发
selector.select(waitTimeOut);
Set<SelectionKey> selected;
synchronized (this) {
// 获取IO事件保定的SelectionKey对象
selected = selector.selectedKeys();
}
// 更新now属性为当前时间戳
updateNow();
for (SelectionKey k : selected) {
SocketChannel sc = ((SocketChannel) k.channel());
// 先判断SelectionKey事件是否是连接事件
if ((k.readyOps() & SelectionKey.OP_CONNECT) != 0) {
// 如果是连接事件，则调用finishConnect()确保已连接成功
if (sc.finishConnect()) {
// 连接成功后更新发送时间
updateLastSendAndHeard();
// 执行主要的连接方法，准备发送ZK的连接请求
sendThread.primeConnection();
}
} else if ((k.readyOps() &
(SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
// 再判断是否是OP_READ或者OP_WRITE事件
// 如果满足则调用doIO方法来处理对应的事件，doIO便是处理获取的
// IO事件核心方法
doIO(pendingQueue, outgoingQueue, cnxn);
}
}
// 执行到这里说明本次触发的NIO事件已经全部执行完毕，但是有可能在途中会
// 产生新的NIO事件需要执行，因此这里会判断是否有可发送的Packet包，如果有
// 则开启OP_WRITE操作，以方便下次直接发送
if (sendThread.getZkState().isConnected()) {
synchronized(outgoingQueue) {
// 查看是否有可发送的Packet包数据
if (findSendablePacket(outgoingQueue, cnxn.sendThread
.clientTunneledAuthenticationInProgress())!=null) {
// 打开OP_WRITE操作
enableWrite();
}
}
}
// 清除SelectionKey集合
selected.clear();
}
void doIO(List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue,
ClientCnxn cnxn) throws InterruptedException, IOException {
SocketChannel sock = (SocketChannel) sockKey.channel();
if (sock == null) {
throw new IOException("Socket is null!");
}
// 这里有处理OP_READ类型的判断，即处理ZK的Server端传过来的请求
// 在第一步中不会走到这里面去，因此忽略
if (sockKey.isReadable()) {
...
}
// 处理OP_WRITE类型事件，即处理要发送到ZK的Server端请求包数据
if (sockKey.isWritable()) {
// 保证线程安全
synchronized(outgoingQueue) {
// 获取最新的需要发送的数据包，这里获取的便是前面SendThread
// 放进去的只有ConnectRequest的Packet包对象
Packet p = findSendablePacket(outgoingQueue, cnxn
.sendThread.clientTunneledAuthenticationInProgress());
if (p != null) {
// 更新最后的发送时间
updateLastSend();
// 如果Packet包的ByteBuffer为空则调用createBB()创建
// 连接时ByteBuffer是一定为空的，因此这里会一定进入
if (p.bb == null) {
if ((p.requestHeader != null) &&
(p.requestHeader.getType() != OpCode.ping) &&
(p.requestHeader.getType() != OpCode.auth)) {
p.requestHeader.setXid(cnxn.getXid());
}
// createBB方法的作用便是序列化请求并将byte[]数组
// 添加到ByteBuffer中
p.createBB();
}
// 使用获取的SocketChannel写入含有序列化数据的ByteBuffer
sock.write(p.bb);
if (!p.bb.hasRemaining()) {
// 发送成功并删除第一个Packet包对象
sentCount++;
outgoingQueue.removeFirstOccurrence(p);
// 如果requestHeader不为空，不是ping或者auth类型的
// 则将Packet包对象添加到pendingQueue中，代表这个
// 包对象正在被Server端处理且没有响应回来
// （需要注意的是只有连接时的ConnectRequest请求头
// requestHeader才会为空，因此这里的条件便是除了
// 新建连接、ping和auth类型的，其它都会被添加进来）
if (p.requestHeader != null
&& p.requestHeader.getType() != OpCode.ping
&& p.requestHeader.getType() != OpCode.auth) {
synchronized (pendingQueue) {
pendingQueue.add(p);
}
}
}
}
// 如果outgoingQueue为空或者尚未连接成功且本次的Packet包对象
// 已经发送完毕则关闭OP_WRITE操作，因此发送ConnectReuqest请
// 求后便需要等待Server端的相应确认建立连接，不允许Client端
// 这边主动发送NIO信息
if (outgoingQueue.isEmpty()) {
disableWrite();
} else if (!initialized && p != null &&
!p.bb.hasRemaining()) {
disableWrite();
} else {
// 为了以防万一打开OP_WRITE操作
enableWrite();
}
}
}
}
private Packet findSendablePacket(LinkedList<Packet> outgoingQueue,
boolean clientTunneledAuthenticationInProgress) {
synchronized (outgoingQueue) {
// 判断outgoingQueue是否为空
if (outgoingQueue.isEmpty()) {
return null;
}
// 两种条件：
// 如果第一个的ByteBuffer不为空
// 如果传入进来的clientTunneledAuthenticationInProgress为false
// 参数为false说明认证尚未配置或者尚未完成
if (outgoingQueue.getFirst().bb != null
|| !clientTunneledAuthenticationInProgress) {
return outgoingQueue.getFirst();
}
// 跑到这里说明认证已完成，需要遍历outgoingQueue数组，把连接的
// 请求找到并放到队列的第一个，以保证下次读取会读取到连接请求
ListIterator<Packet> iter = outgoingQueue.listIterator();
while (iter.hasNext()) {
Packet p = iter.next();
// 只有连接的requestHeader是空的，因此只需要判断这个条件即可
// 其它类型的包数据header肯定是不为空的
if (p.requestHeader == null) {
// 先删除本包，随后放到第一位
iter.remove();
outgoingQueue.add(0, p);
return p;
}
}
// 执行到这里说明确实没有包需要发送
return null;
}
}
}当Socket把请求数据已经序列化到ByteBuffer中的数据发出去后，Client端的第一步便已经完成。从这个流程中最关键的就是把OP_READ操作看成接收Server端的响应，而OP_WRITE则是Client主动发数据和Server端进行交互的操作，这样在看代码理解时会更加轻松。
2.5 客户端启动流程图
由于篇幅受限，剩下的两个步骤（Server接收处理及响应和Client端接收Server端响应）在下篇文章分析，敬请期待。