ZooKeeper客户端源码——向服务端建立连接+会话建立+心跳保持长连接 _zookeeper

得意犹堪夸世俗，诏黄新湿字如鸦。这篇文章主要讲述ZooKeeper客户端源码——向服务端建立连接+会话建立+心跳保持长连接相关的知识，希望能为你提供帮助。

文章图片

从ZooKeeper实例初始化开始ZooKeeper 提供了原生的客户端库，虽然不好用，但是能够更好理解客户端与服务端建立连接和通信的过程。比较流行的Apache Curator也是对原生库的再封装。
向服务端建立连接，只需要实例化一个ZooKeeper对象，将服务器地址列表传进去即可。因为发起连接请求是一个异步的过程，所以实例化ZooKeeper时可以传一个Watcher，会话建立成功之后，客户端会生成一个 “已经建立连接（SyncConnected）” 的事件，进行回调通知。只有会话建立成功之后，才能与服务端进行通信。

String connectString = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183"; ZooKeeper zooKeeper = new ZooKeeper(connectString, 20000, new Watcher() @Override public void process(WatchedEvent event) if (event.getState() == Event.KeeperState.SyncConnected) System.out.println("会话建立成功"); );

ZooKeeper实例初始化流程如下：
【ZooKeeper客户端源码——向服务端建立连接+会话建立+心跳保持长连接】

文章图片

创建HostProvider
HostProvider顾名思义就是“服务地址提供器”，默认实现类StaticHostProvider。StaticHostProvider核心思想是，将服务地址列表打乱，然后构成一个虚拟环，轮询向外提供服务地址。
如果StaticHostProvider不满足需求，可以自定义实现HostProvider。
打乱服务地址列表打乱地址就很简单了，用了java官方提供的工具java.util.Collections#shuffle。

// org.apache.zookeeper.client.StaticHostProvider#shuffle private List< InetSocketAddress> shuffle(Collection< InetSocketAddress> serverAddresses) List< InetSocketAddress> tmpList = new ArrayList< > (serverAddresses.size()); tmpList.addAll(serverAddresses); Collections.shuffle(tmpList, sourceOfRandomness); return tmpList;

随机源生成：

Random sourceOfRandomness = new Random(System.currentTimeMillis() ^ this.hashCode())

随机源的种子是当前时间毫秒值掺杂（^）当前StaticHostProvider实例的hashCode，使得每次生成的随机源更公平。
构建虚拟环轮询负载如何将一个服务地址列表构成一个环轮询负载的呢？
有两个游标，currentIndex 和 lastIndex是实现“虚拟环轮询负载”的关键：

currentIndex，指向当前选择的位置。每选择一次就加一，如果等于服务地址列表长度，就重置为0，这样就形成了一个环。
lastIndex，上次选择的位置。会话建立成功之后会将currentIndex赋值给lastIndex。

public InetSocketAddress next(long spinDelay) boolean needToSleep = false; InetSocketAddress addr; synchronized (this) // 省略部分无关代码// currentIndex自增，如果等于服务地址列表长度，就重置为0 ++currentIndex; if (currentIndex == serverAddresses.size()) currentIndex = 0; addr = serverAddresses.get(currentIndex); // 两个游标currentIndex、lastIndex // currentIndex 当前选择的位置，lastIndex上次选择的位置 // lastIndex 什么时候设置呢？会话建立成功之后调用 onConnected，将currentIndex赋值给lastIndex needToSleep = needToSleep || (currentIndex == lastIndex & & spinDelay > 0); if (lastIndex == -1) lastIndex = 0; // 如果 currentIndex和lastIndex且spinDelay> 0，就需要休眠spinDelay时间， if (needToSleep) try Thread.sleep(spinDelay); catch (InterruptedException e) LOG.warn("Unexpected exception", e); // 解析InetSocketAddress， // 如果一个主机映射了多个ip地址（InetAddress） // 就打乱选择其中一个地址返回 return resolve(addr);

当currentIndex和lastIndex相等时，且

spinDelay&
gt;
0

，就会休眠spinDelay毫秒，然后再将选择的服务地址返回，为什么会有这个逻辑呢？
这是因为，如果轮询了一圈服务地址都没有成功建立连接，与其一味不停地重试，还不如休眠一段时间再试，可能成功的概率更高一些。
创建ConnectStringParser
ConnectStringParser 就是将 connectString 按一定格式解析成 InetSocketAddress 列表。connectString格式如下：
如果在服务地址后面指定了路径，后续的操作都是在该路径下进行。

public ConnectStringParser(String connectString) // parse out chroot, if any // 解析chroot // connectString 可以指定某个路径， // 如127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183/test int off = connectString.indexOf(/); if (off > = 0) String chrootPath = connectString.substring(off); // ignore "/" chroot spec, same as null if (chrootPath.length() == 1) this.chrootPath = null; else PathUtils.validatePath(chrootPath); this.chrootPath = chrootPath; connectString = connectString.substring(0, off); else this.chrootPath = null; // 按 , 分割 List< String> hostsList = split(connectString, ","); for (String host : hostsList) int port = DEFAULT_PORT; String[] hostAndPort = NetUtils.getIPV6HostAndPort(host); if (hostAndPort.length != 0) host = hostAndPort[0]; if (hostAndPort.length == 2) port = Integer.parseInt(hostAndPort[1]); else int pidx = host.lastIndexOf(:); if (pidx > = 0) // otherwise : is at the end of the string, ignore if (pidx < host.length() - 1) port = Integer.parseInt(host.substring(pidx + 1)); host = host.substring(0, pidx); // 封装未解析的InetSocketAddress serverAddresses.add(InetSocketAddress.createUnresolved(host, port));

创建并启动ClientCnxn
ClientCnxn是对客户端连接的抽象和封装，负责网络连接管理和watcher管理。
还记得从ZooKeeper构造器传入的Watcher吗？它会作为默认Watcher传给ZKWatchManager，后续其他请求注册watcher时，可以不用再定义，直接使用默认的Watcher。
初始化SendThread时会传入一个ClientCnxnSocket，ClientCnxnSocket是对网络底层的封装，默认实现类为ClientCnxnSocketNIO。

文章图片

ClientCnxn创建好后会进行启动操作，就是启动SendThread和EventThread两个线程。后续的网络连接建立和通信都是由SendThread线程负责。
向服务端建立连接SendThread启动后，会进入一个循环状态，首先判断是否已经建立连接，如果没有就通过hostProvider选择一个服务地址发起连接。

文章图片

网络底层处理类ClientCnxnSocket，以ClientCnxnSocketNIO实现为准，如下图是建立连接的过程：

文章图片

ClientCnxnSocketNIO底层是创建了非阻塞的SocketChannel，然后注册OP_CONNECT事件，并发起连接，如果此时能立刻连上，就继续进行会话建立流程。

文章图片

如下模拟服务器一直连不上，多次通过hostProvider轮询选择服务器重连的效果：
服务器地址选择了一圈以后，会休眠spinDelay毫秒，也符合源码逻辑。

文章图片

会话建立请求连接建立之后，需要继续进行会话建立请求，因为每条连接都是有状态的，临时节点的归属就是以会话为依托，连接断开，会话失效，临时节点也就跟随着清除。
每条连接跟会话绑定，如果连接因为网络等问题断开，在会话有效期内重连上，就可以恢复之前的工作场景，而如果在会话失效之后重连上，就是非法连接，需要重新进行会话建立。

文章图片

发起会话建立请求
发起会话建立，首先构建一个ConnectRequest请求体：

ConnectRequest conReq = new ConnectRequest(0, lastZxid, sessionTimeout, sessId, sessionPasswd);

ConnectRequest需要传入客户端最近一次的事务zxid，会话超时时间，会话id，会话密码。首次建立会话，sessionId为0，sessionPasswd为空。
其次，将ConnectRequest请求体包装进Packet对象，Packet是ZooKeeper通信的最小单元，所有请求体都会包装进一个Packet对象再序列化发送给服务端。

Packet(RequestHeader requestHeader,ReplyHeader replyHeader,Record request, Record response, WatchRegistration watchRegistration, boolean readOnly)

对于ConnectRequest包装的Packet，请求头RequestHeader为null，并且会被放在请求队列outgoingQueue的首位。

outgoingQueue.addFirst(new Packet(null, null, conReq, null, null, readOnly));

接着注册网络IO读写事件，后续请求包就会被发送给服务端了，发送完成后再将会话建立的Packet从outgoingQueue中移除。

文章图片

注：发送到服务端的会话创建流程，暂时不需要了解，后面讲解服务端源码时会详细讲述。
会话建立响应
会话建立响应，需要和普通请求响应分开处理，如果接收到响应信息，判断客户端还未初始化完成，就认为这个响应一定是会话建立响应。

文章图片

首先从底层网络缓冲区读取数据反序列化构建ConnectResponse响应体，解析出经过服务器协商好的sessionTimeout以及sessionId 和 sessionPasswd。

文章图片

根据协商的sessionTimeout重新设置readTimeout和connectTimeout，readTimeout是negotiatedSessionTimeout的2/3，connectTimeout是协商negotiatedSessionTimeout与服务地址列表个数平均。

文章图片

最后生成一个SyncConnected的watcher事件，交由EventThread线程进行回调，这是唯一一个不需要向服务端注册的watcher事件，完全由客户端自己生成和触发。
心跳保持长连接网络建立连接，会话建立都完成后，就可以与服务端通信了，为了保持长连接的会话一直有效，在没有向服务端发送请求的一段时间内会发送心跳请求。而一段时间是多久？如下是计算下一次发送心跳请求时间的算法：

// clientCnxnSocket.getIdleSend为距离上次发送的时长 // readTimeout = sessionTimeout * 2 / 3 int timeToNextPing = readTimeout / 2 - clientCnxnSocket.getIdleSend() - ((clientCnxnSocket.getIdleSend() > 1000) ? 1000 : 0);

根据计算，每隔sessionTimeout/3如果没有发送任何请求，就发送一次心跳。多减1秒是为了防止因为竞态情况而丢失心跳请求。
发送的心跳请求数据很简单，没有请求体，只有请求头：

文章图片

如下是在源码中加了日志后的心跳过程：
协商sessionTimeout为9999，readTimeout计算得6666，timeToNextPing为3333，每次会多减1秒，大概每隔3秒发一次心跳。

文章图片

总结

客户端与服务端建立的是长连接，如果连接失败，服务地址列表会轮询重试，直到连接成功，官方提供了默认的服务地址负载算法 StaticHostProvider，也可以自己实现。
每条连接是有状态的，只有建立了会话，才能真正开始与服务端通信。会话建立成功之后，会生成一个SyncConnected事件进行回调通知。
会话是临时节点的基础，在会话有效期内断开重连，可以恢复上一次工作场景。
为了保持长连接的会话一直有效，在没有向服务端发送请求的一段时间内会发送心跳请求，心跳间隔时间为sessionTimeout/3。

如若文章有错误理解，欢迎批评指正，同时非常期待你的评论、点赞和收藏。