前言
ZooKeeper是一个相对简单的分布式协调服务，通过阅读源码我们能够更进一步的清楚分布式的原理。
环境
ZooKeeper 3.4.9
入口函数
在bin/zkCli.sh中，我们看到client端的真实入口其实是一个org.apache.zookeeper.ZooKeeperMain的Java类
"$JAVA" "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}" \
-cp "$CLASSPATH" $CLIENT_JVMFLAGS $JVMFLAGS \
org.apache.zookeeper.ZooKeeperMain "$@"
通过源码走读，看到在ZooKeeperMain中主要由两部分构成
connectToZK(cl.getOption("server"));
while ((line = (String)readLine.invoke(console, getPrompt())) != null) {
executeLine(line);
}
- 构造一个
ZooKeeper对象，同ZooKeeperServer进行建立通信连接 - 通过反射调用
jline.ConsoleReader类，对终端输入进行读取，然后通过解析单行命令，调用ZooKeeper接口。
如上所述，client端其实是对 zookeeper.jar 的简单封装，在构造出一个ZooKeeper对象后，通过解析用户输入，调用 ZooKeeper 接口和 Server 进行交互。
ZooKeeper 类
刚才我们看到 client 端同 ZooKeeper Server 之间的交互其实是通过 ZooKeeper 对象进行的，接下来我们详细深入到 ZooKeeper 类中，看看其和服务端的交互逻辑。
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
boolean canBeReadOnly)
throws IOException
{
ConnectStringParser connectStringParser = new ConnectStringParser(connectString);
HostProvider hostProvider = new StaticHostProvider( connectStringParser.getServerAddresses());
cnxn = new ClientCnxn(connectStringParser.getChrootPath(),
hostProvider, sessionTimeout, this, watchManager, getClientCnxnSocket(), canBeReadOnly);
cnxn.start();
}
在 ZooKeeper的构造方法中,可以看到 ZooKeeper 中使用 Server 的服务器地址构建了一个 ClientCnxn 类，在这个类中，系统新建了两个线程
sendThread = new SendThread(clientCnxnSocket);
eventThread = new EventThread();
其中，SendThread 负责将ZooKeeper的请求信息封装成一个Packet，发送给 Server ,并维持同Server的心跳，EventThread负责解析通过通过SendThread得到的Response，之后发送给Watcher::processEvent进行详细的事件处理。
Client 时序图
如上图所示，Client中在终端输入指令后，会被封装成一个Request请求，通过submitRequest，进一步被封装成Packet包，提交给SendThread处理。
SendThread通过doTransport将Packet发送给Server,并通过readResponse获取结果，解析成一个Event，再将Event加入EventThread的队列中等待执行。
EventThread通过processEvent消费队列中的Event事件。
SendThread
SendThread 的主要作用除了将Packet包发送给Server之外，还负责维持Client和Server之间的心跳，确保 session 存活。
现在让我们从源码出发，看看SendThread究竟是如何运行的。
SendThread是一个线程类，因此我们进入其run()方法，看看他的启动流程。
while (state.isAlive()) {
if (!clientCnxnSocket.isConnected()) {
// 启动和server的socket链接
startConnect();
}
// 根据上次的连接时间，判断是否超时
if (state.isConnected()) {
to = readTimeout - clientCnxnSocket.getIdleRecv();
} else {
to = connectTimeout - clientCnxnSocket.getIdleRecv();
}
if (to <= 0) {
throw new SessionTimeoutException(warnInfo);
}
// 发送心跳包
if (state.isConnected()) {
if (timeToNextPing <= 0 || clientCnxnSocket.getIdleSend() > MAX_SEND_PING_INTERVAL) {
sendPing();
clientCnxnSocket.updateLastSend();
}
}
// 将指令信息发送给 Server
clientCnxnSocket.doTransport(to, pendingQueue, outgoingQueue, ClientCnxn.this);
}
从上面的代码中，可以看出SendThread的主要任务如下:
- 创建同 Server 之间的 socket 链接
- 判断链接是否超时
- 定时发送心跳任务
- 将ZooKeeper指令发送给Server
与 Server 的长链接
ZooKeeper通过获取ZOOKEEPER_CLIENT_CNXN_SOCKET变量构造了一个ClientCnxnSocket对象，默认情况下是ClientCnxnSocketNIO类
String clientCnxnSocketName = System
.getProperty(ZOOKEEPER_CLIENT_CNXN_SOCKET);
if (clientCnxnSocketName == null) {
clientCnxnSocketName = ClientCnxnSocketNIO.class.getName();
}
在ClientCnxnSocketNIO::connect中我们可以看到这里同Server之间创建了一个socket链接。
SocketChannel sock = createSock();
registerAndConnect(sock, addr);
超时与心跳
在SendThread::run中，可以看到针对链接是否建立分别有readTimeout和connetTimeout 两种超时时间，一旦发现链接超时，则抛出异常，终止 SendThread。
在没有超时的情况下，如果判断距离上次心跳时间超过了1/2个超时时间，会再次发送心跳数据，避免访问超时。
发送 ZooKeeper 指令
在时序图中，我们看到从终端输入指令后，我们会将其解析成一个Packet 包，等待SendThread进行发送。
以ZooKeeper::create为例
RequestHeader h = new RequestHeader();
h.setType(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);
在这里create指令，被封装成了一个 CreateRequest，通过submitRequest被转成了一个Packet包
public ReplyHeader submitRequest(RequestHeader h, Record request,
Record response, WatchRegistration watchRegistration)
throws InterruptedException {
ReplyHeader r = new ReplyHeader();
Packet packet = queuePacket(h, r, request, response, null, null, null, null, watchRegistration);
synchronized (packet) {
while (!packet.finished) {
packet.wait();
}
}
return r;
}
Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
Record response, AsyncCallback cb, String clientPath,
String serverPath, Object ctx, WatchRegistration watchRegistration) {
Packet packet = null;
// Note that we do not generate the Xid for the packet yet. It is
// generated later at send-time, by an implementation of ClientCnxnSocket::doIO(),
// where the packet is actually sent.
synchronized (outgoingQueue) {
packet = new Packet(h, r, request, response, watchRegistration);
packet.cb = cb;
packet.ctx = ctx;
packet.clientPath = clientPath;
packet.serverPath = serverPath;
if (!state.isAlive() || closing) {
conLossPacket(packet);
} else {
// If the client is asking to close the session then
// mark as closing
if (h.getType() == OpCode.closeSession) {
closing = true;
}
outgoingQueue.add(packet);
}
}
sendThread.getClientCnxnSocket().wakeupCnxn();
return packet;
}
在submitRequest中，我们进一步看到Request被封装成一个Packet包，并加入SendThread::outgoingQueue队列中，等待执行。
Note:在这里我们还看到，ZooKeeper方法中所谓的同步方法其实就是在Packet被提交到SendThread之后，陷入一个while循环，等待处理完成后再跳出的过程
在SendThread::run的while循环中，ZooKeeper通过doTransport将存放在outgoingQueue中的Packet包发送给 Server。
void doIO(List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue, ClientCnxn cnxn) {
if (sockKey.isReadable()) {
// 读取response信息
sendThread.readResponse(incomingBuffer);
}
if (sockKey.isWritable()) {
Packet p = findSendablePacket(outgoingQueue, cnxn.sendThread.clientTunneledAuthenticationInProgress());
sock.write(p.bb);
}
}
在doIO发送socket信息之前，先从socket中获取返回数据，通过readResonse进行处理。
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() == -1) {
WatcherEvent event = new WatcherEvent();
event.deserialize(bbia, "response");
WatchedEvent we = new WatchedEvent(event);
eventThread.queueEvent( we );
}
}
在readReponse中,通过解析数据，我们可以得到WatchedEvent对象，并将其压入EventThread的消息队列，等待分发
EventThread
public void run() {
while (true) {
Object event = waitingEvents.take();
if (event == eventOfDeath) {
wasKilled = true;
} else {
processEvent(event);
}
}
在EventThread中通过processEvent对队列中的事件进行消费，并分发给不同的Watcher
watch事件注册和分发
通常在ZooKeeper中，我们会为指定节点添加一个Watcher，用于监听节点变化情况，以ZooKeeper:exist为例
// the watch contains the un-chroot path
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();
ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
代码的大致逻辑和create类似，但是对wathcer做了一层ExistWatchRegistration的包装，当packet对象完成请求之后，调用register方法，根据不同包装的WatchRegistration将watch注册到不同watch列表中，等待回调。
if (p.watchRegistration != null) {
p.watchRegistration.register(p.replyHeader.getErr());
}
在 ZooKeeper 中一共有三种类型的WatchRegistration，分别对应DataWatchRegistration,ChildWatchRegistration,ExistWatchRegistration。 并在ZKWatchManager类中根据每种类型的WatchRegistration,分别有一张map表负责存放。
private final Map<String, Set<Watcher>> dataWatches =
new HashMap<String, Set<Watcher>>();
private final Map<String, Set<Watcher>> existWatches =
new HashMap<String, Set<Watcher>>();
private final Map<String, Set<Watcher>> childWatches =
new HashMap<String, Set<Watcher>>();
当 EventThread::processEvent 时，根据event的所属路径，从三张map中获取对应的watch列表进行消息通知及处理。
总结
client 端的源码分析就到此为止了。
ZooKeeper Client 的源码很简单，拥有三个独立线程分别对命令进行处理，分发和响应操作，在保证各个线程相互独立的基础上，尽可能避免了多线程操作中出现锁的情况。
网友评论