zookeeper客户端与服务端交互流程源码解析

78 阅读 0 评论 52 点赞

我是靠谱客的博主舒心溪流，最近开发中收集的这篇文章主要介绍zookeeper客户端与服务端交互流程源码解析，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

客户端发送请求和接收请求。

图1

图1 zkCll.sh 启动时会调用zookeeperMain() 方法。古这是入口。

图2

图2 290行 new ZookeeperMain , 297行连接zk

图3

图 3 277 判断zk的状态， 281行设置只读， 282行 new Zookeeper（）

图4

图4 445 行赋值watcher给默认的defaultWatcher. 447行对地址进行处理， 447行包装地址。

图5

图5 如果写多个地址的话zookeeper回随机选择一个地址连接，如果这个地址挂了，他会从其他的地址再从新选一个连接。

图6

图7

图8

图6 图7 图8 意思是传进来的地址以逗号截取，然后每一个地址再以":" 分号截取，封装到inetSocketAddressHolder类中，然后把 inetSocketAddressHolder 放在一个ArrayList中serverAddresses中。

图9

图10

图 9 图10没有做jut具体的事情，104行打乱的意思（图10），换句话说就是随机地址连接。

图10

图11

ClientCnxSockNIO

图12

图11 图12 1884行先拿图12 的key 去配置文件中查找是否配置这个名字，如果没有配置ClientCnxSockNIO 的全类名，然后反射实例化返回。

图13

图13初始化Zookeeper参数值，创建两个线程。

图14

图15

图16

图14,图15，图16 启动两个线程，SendThread 继承ZookeeperThread 这个线程。Zookeerper线程继承Thread线程。

图17

图17-0

17-1

图17-1

图17-2

图17-3

图17-4

图17-5

图17-6

图17-7

图17-8

图17-9

图17-9 80行如果还没有初始化，等待连接返回的结果图17-12。92行如果初始化完了，改为true 。94行客户端发送后，如何响应呢，94行sendThread.readRespone() 用这个线程读取响应结果图17-15。

图17-10

图17-11

图17-12

图17-12核心方法为146行调用另一个线程的onConnected的方法图17-13

图17-13

图17-13 1312行如果连接成功本地注册一个Watcher的None事件。

图17-14

图17-15

图17-15 通过判断xid的值来分别处理，-2 ping的处理，-4验证，-1 客户端接收服务的的事件。 823行 packet的replyHeader的xid和replyHdr的xid一致时，把replyHdr的请求头信息从新封装到packet中，842行中，如果服务端返回信息，同样把新增加的服务端返回信息封装到 packet中. replyHdr是什么呢,是把从参数传入的流信息进行解析转换为replyHdr,此时这个replyHdr没有response信息。850行完成通知图17-16

图17-16

图17-16 完成唤醒操作。

图17 为SendThread 的run()方法。图17 图18 1044行默认状态为未连接状态。如果不是关闭状态和验证失败状态就是活着的状态图19 ；1046行如果未连接，是第一次连接1047行的if判断. 会走1062行，从hostPovider选取一个地址，进行socket连接。 1100行为读取超时= 设置的超时时间减去当前时间距离上次读取时间（now -上次读取时间）<0 则认为是超时，或者，设置的连接事件减去当前时间距离上次发送时间（now -上次发送时间）小于0认为连接超时则抛出异常sessionTimeOutException。然后try catch捕获异常，程序不会中止，继续执行while循环,1044行；当第一次connect()方法时图17-1，图17-2，图17-3 方法把isFirstConnect置为false,当抛出sessionTimeOutException 异常时，进入1047行到1053行睡眠一段时间。1062行，图17-4 默认超时情况下从新从其他的地址中选取一个地址，不会是上一次执行的地址，因为会采用取余数的方式，图17-4；图1115行连接成功，就返送ping.图17-5 ，和发送逻辑一致，只不过类型为OpCode.ping ;xid=-2. ； 1115行连接成功后1145行客户端发送数据

图17 到图17-14 总结；首先1.初始化 SendThread线程 ,初始化 socket ； 2.zookeeper 创建一条命令，然后这个命令加到 request ，把Request 变成Packet ，最后放在outgoingQueue队列中。sendThread 线程中有一个while(){ 1.如果socket没有连接就去连接，2，如果socket连接成功了，客户端会发送一个ConnectRequest, 接受服务端返回的ConnectResponse(Envent.none) 3,如果连接成功，从outgoingQueue 队列中取出packet ,通过socket发送出去，同时，如果说有结果。penedingqueue里面回去放等待结果的packet}

图18

图19

图20

图20 995行改变状态为正在连接，conecting . 设置名字。999行到1018行不重要，1019行打印日志不重要。 1021行socket连接。

图21

图22

图23

图24

图23 连接socket ClientCnxnSocket 接口的 connect() 方法，他有唯一的实现类ClientCnxnSocketNIO的connect()方法图24.

首先创建一个socket图25. 287行注册一个socket图26.

图25

图26

图26 276行注册一个连接事件，277行判断是不是立即连接成功，如果是调用primeConnection()方法。

图27

总结；

图28-0

图28-0 290行为执行完成以后，执行291行命令行的run（）方法。

图28

图28 314行，316行为java 的命令行实现类。332行为核心的实现类方法图31。执行命令图30命令行前缀对应的代码图29；

图29

图30

图31

图31 如果命令不为空，加入命令，加入历史，处理命令。命令次数加1.重点是372行。

图32

图33

图33 689行创建命令如果有-e,-s 为创建临时有序节点，如果 -e 就是临时节点，-s就是持久节点顺序 704行返回权限列表图34 ，707行调用zk 创建节点。709行打印命令。649行推出命令调用了zk.close()方法图 36

图34

图35

图36

图37

图37 777行验证path ，传一个是否顺序节点标志。如果是顺序节点会在path上固定加上+1图38 ；779行创建一个根目录图40；781行构建一个请求头图41，设置一个类型为create 图42 ，构建一个createRequest ， createResponse请求，设置内容，标记，path，权限，792行 submitRequest 图43，通过socket 发送请求到服务端，获取相应头，判断相应头是否有错误信息，有抛出异常，如果没有,拿到path返回。

图38