Zookeeper之客户端连接源码分析

image.png

客户端连接

通过Zookeeper客户端类库连接org.apache.zookeeper.ZooKeeper

ZooKeeper zooKeeper = new ZooKeeper("127.0.0.1:2181", 100*1000, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("conn");
            }
        });

1、ZooKeeper类的创建
org.apache.zookeeper.ZooKeeper#ZooKeeper

image.png

通过createDefaultHostProvider创建静态服务提供类StaticHostProvider

image.png

2、ClientCnxn类的创建

image.png

org.apache.zookeeper.ZooKeeper#createConnection

image.png

其中最重要的是watchManager、sendThread、eventThread的初始化
3、ClientCnxn类的启动
ClientCnxn类的启动便是sendThread和eventThread线程的启动

image.png

在具体分析两个线程之前，需要了解pendingQueue和outgoingQueue队列表示的意义。
outgoingQueue：表示需要发送的数据包
pendingQueue：表示已经发送正在等待响应的数据包

SendThread

1、run
run方法内部有一个while方法，只要不为false将会一直运行
org.apache.zookeeper.ClientCnxn.SendThread#run

image.png

org.apache.zookeeper.ZooKeeper.States#isAlive
如果state状态不是CLOSED或AUTH_FAILED将会一直循环

image.png

2、建立socket连接
如果clientCnxnSocket客户端socket没有连接即sockKey = null，将会调用连接方法

image.png

org.apache.zookeeper.ClientCnxn.SendThread#startConnect

image.png

org.apache.zookeeper.ClientCnxnSocketNIO#connect

image.png

①、创建socket对象
org.apache.zookeeper.ClientCnxnSocketNIO#createSock

image.png

②、注册连接事件
org.apache.zookeeper.ClientCnxnSocketNIO#registerAndConnect

image.png

再尝试连接，如果immediateConnect返回true则会调用primeConnection方法。这里连接失败返回fasle
③、incomingBuffer和lenBuffer缓存重置

image.png

3、AuthFailed

image.png

把eventOfDeath添加到waitingEvents

image.png

4、发送心跳ping

image.png

org.apache.zookeeper.ClientCnxn.SendThread#sendPing

image.png

即是把ping包添加到outgoingQueue
5、只读模式，寻找读/写服务器

image.png

6、doTransport

image.png

org.apache.zookeeper.ClientCnxnSocketNIO#doTransport
selector.select查询有哪些事件。如果没有事件便会阻塞waitTimeOut时间

image.png

6.1、连接事件准备就绪
如果是连接事件准备就绪，则会调用primeConnection方法

image.png

org.apache.zookeeper.ClientCnxn.SendThread#primeConnection
用来设置会话session，watches和authentication
①、watch

image.png

用来判断这些dataWatches、existWatches、childWatches、persistentWatches、persistentRecursiveWatches等watch是否为空

image.png

用来判断persistentWatches持久化的watch与persistentRecursiveWatches持久化的递归Watche是否为空。
最后只要任意watch不为空，都会添加到outgoingQueue队列中

image.png

②、authInfo
添加权限的authInfo请求数据包

image.png

③、添加空的数据包

image.png

④、更改感兴趣的事件类型

image.png

org.apache.zookeeper.ClientCnxnSocketNIO#doIO
6.2、处理读写事件

image.png

①、读事件(得到服务端响应数据)

image.png

首先读取4字节长度的数据到incomingBuffer
org.apache.zookeeper.ClientCnxnSocket#readLength
把incomingBuffer设置为还要读取的长度

image.png

最后通过readResponse读取响应结果

image.png

org.apache.zookeeper.ClientCnxn.SendThread#readResponse

image.png

包括PING_XID(pings)、AUTHPACKET_XID(AuthPacket)、WATCHER_EVENT(watch事件)

image.png

finishPacket

image.png

org.apache.zookeeper.ClientCnxn#finishPacket

image.png

org.apache.zookeeper.ClientCnxn.EventThread#queuePacket

image.png

②、写事件(发送数据到服务端)

image.png

a、findSendablePacket
从outgoingQueue队列中获取第一个并封装成Packet数据包，对于关于authentication权限的数据包，只发送header非空的数据

image.png

b、createBB缓存数据
org.apache.zookeeper.ClientCnxn.Packet#createBB
设置数据包Packet.ByteBuffer bb缓存数据

image.png

c、通过socker网络传输到服务端
对于发送过的数据并且不是ping或auth的数据包要放到pendingQueue队列中等待响应

image.png

d、禁止写数据
如果outgoingQueue是空或没有初始化将会禁止写数据(即禁止向服务端发送数据)

image.png

outgoingQueue队列什么时候添加

提交请求时，会把当前数据包添加outgoingQueue
1、getData
org.apache.zookeeper.ZooKeeper#getData

image.png

org.apache.zookeeper.ClientCnxn#submitRequest

image.png

org.apache.zookeeper.ClientCnxn#queuePacket

image.png

2、create

org.apache.zookeeper.ZooKeeper#create

image.png

也会调用到submitRequest方法
所以，只要最终调用到queuePacket方法都会添加，zookeeper客户端的增删改查等操作都会添加到outgoingQueue队列中

EventThread

从waitingEvents队列中获取响应数据，阻塞读取数据
org.apache.zookeeper.ClientCnxn.EventThread#run

image.png

org.apache.zookeeper.ClientCnxn.EventThread#processEvent
1、watcher
调用watcher.process方法

image.png

2、LocalCallback
调用异步执行方法AsyncCallback.processResult方法（LocalCallback事件）

image.png

3、response
根据response的类型调用异步执行方法

image.png

总结：

客户端的连接通过ZooKeeper创建，创建ZooKeeper类将会创建并启动ClientCnxn(客户端上下文类)。创建ClientCnxn类会引起sendThread、sendThread线程的初始化和启动。
SendThread：创建socket连接、获取命令数据发送给服务端(ping、auth、创建删除节点)、读取服务端响应数据
EventThread：从waitingEvents队列中获取数据。执行监听器watch事件、执行Packet数据包的异步调用