Zookeeper 源码入门

首先分享之前的所有文章 , 欢迎点赞收藏转发三连下次一定 >>>> 😜😜😜

文章合集 : 🎁 juejin.cn/post/694164…

Github : 👉 github.com/black-ant

CASE 备份 : 👉 gitee.com/antblack/ca…

一. 前言

出于对集群选举流程的好奇 , 所以把 Zookeeper 源码拉下来跑了一下 , 这篇文档对这个过程做一个简单的记录.

要想看懂任何源码 , 第一步就是要跑起来 . 这一篇主要介绍 , 如果快速的跑源码 ,同时简单介绍其中的关键点 , 便于处理

二 . 源码的运行

2.1 主启动流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
java复制代码// Step 1 : Git 拉取
https://github.com/apache/zookeeper.git

// Step 2 : 本地运行 (Intellij)
1. 找到 zookeeper-server 子模块
2. 找到 对应的启动类
3. 通过命令启动项目

// PS : 这里查看了 Zookeeper 运行包 (ZkServer.cmd)的内容 , 决定先采用相同的方式启动 >>>>
setlocal
call "%~dp0zkEnv.cmd"

set ZOOMAIN=org.apache.zookeeper.server.quorum.QuorumPeerMain
echo on
call %JAVA% "-Dzookeeper.log.dir=%ZOO_LOG_DIR%" "-Dzookeeper.root.logger=%ZOO_LOG4J_PROP%" -cp "%CLASSPATH%" %ZOOMAIN% "%ZOOCFG%" %*

endlocal

2.2 其他启动类

进入源码可以看到很多其他的启动类 , 这里查阅了一下 API Doc , 大概了解了一下

image.png

2.3 源码的主要逻辑

源码的主要入口类为 QuorumPeerMain , 而其他的类会由 QuorumPeerMain 发起调用 (例如 : ) , 这里提供一个简单的调用流程图 >>>

结构图.jpg

二 . 启动流程

来简单看一下 QuorumPeerMain 启动时做了什么 :

3.1 QuorumPeerMain 简介

当使用该类的main()方法启动程序时,第一个参数被用作配置文件的路径 , 配置文件中可以包含如下信息 :

  • dataDir : ZooKeeper数据所在目录。
  • dataLogDir : ZooKeeper事务日志存放目录。
  • clientPort : 用于与客户端通信的端口。
  • tickTime : 一个滴答的持续时间,单位为毫秒。这是ZooKeeper中的基本时间单位。
  • initLimit : 跟踪者等待与leader初始同步的最大节拍数。
  • syncLimit : 跟踪者等待来自leader的消息(包括心跳)的最大节拍数。
  • server.id : 这是具有给定id的服务器将用于仲裁协议的主机:port[:port]。
1
2
3
4
5
6
java复制代码# 以下是我的配置
tickTime=2000
initLimit=10
syncLimit=5
dataDir=D:\\java\\workspace\\git\\zookeeper\\temp
clientPort=2181

3.2 IDEA 配置

VM Options

1
java复制代码"-Dzookeeper.root.logger=INFO,CONSOLE" -cp "D:\java\workspace\git\zookeeper\zookeeper\zookeeper-server\target\classes;D:\java\workspace\git\zookeeper\zookeeper\zookeeper-server\target\lib\*;D:\java\workspace\git\zookeeper\zookeeper\bin\..\*;D:\java\workspace\git\zookeeper\zookeeper\bin\..\lib\*;D:\java\workspace\git\zookeeper\zookeeper\bin\..\conf"

Program arguments

1
java复制代码D:\java\workspace\git\zookeeper\zoo.cfg

image.png

其中比较重要的就是 log 级别和你的源码路径 , 配置完成后一般项目就能正常跑起来了

三 . 请求与接收

请求和接收跑通了才是一切的基础 , Zk 的核心对象分别为 ClientCnxn 和 ServerCnxn

System-ServerCnxn.png

从 log 中不难发现 , 默认是走的 NIO

1
2
3
4
5
6
java复制代码[main:NIOServerCnxnFactory@89] - binding to port 0.0.0.0/0.0.0.0:2181
[NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxnFactory@222] - Accepted socket connection from /127.0.0.1:53152
[NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:ZooKeeperServer@903] - Connection request from old client /127.0.0.1:53152; will be dropped if server is in r-o mode
[NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:ZooKeeperServer@942] - Client attempting to renew session 0x100001d6f1f0009 at /127.0.0.1:53152
[NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:ZooKeeperServer@687] - Invalid session 0x100001d6f1f0009 for client /127.0.0.1:53152, probably expired
[NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@1056] - Closed socket connection for client /127.0.0.1:53152 which had sessionid 0x100001d6f1f0009

3.1 客户端发起请求

以 SetData 为例 , 主要经过如下流程 :

  • C- ZooKeeper # 相关逻辑
  • C- ClientCnxn # submitRequest : 构建 Packet , 同时加入 Queue 中
  • C- ClientCnxnSocketNIO : 发起处理请求
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
java复制代码public Stat setData(final String path, byte[] data, int version) throws KeeperException, InterruptedException {

final String clientPath = path;
PathUtils.validatePath(clientPath);

// serverPath -> /testWatch
final String serverPath = prependChroot(clientPath);

RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.setData);

// 构建 request
SetDataRequest request = new SetDataRequest();
request.setPath(serverPath);
request.setData(data);
request.setVersion(version);

SetDataResponse response = new SetDataResponse();

// 通过 ClientCnxn 发起请求
ReplyHeader r = cnxn.submitRequest(h, request, response, null);
if (r.getErr() != 0) {
throw KeeperException.create(KeeperException.Code.get(r.getErr()), clientPath);
}
return response.getStat();
}

底层 还是用的 NIO 调用 , 后面详细看看

3.2 服务端处理请求

Zookeeper 的核心管理类为 ZooKeeperServer , 其中包括以下方法 :

  • processConnectRequest : 处理连接请求

核心一 : processConnectRequest

创建连接是一切的起点 , 主要通过以下流程调用到该类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
java复制代码C- ZooKeeperServer : 仅保留核心代码
public void processConnectRequest(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException {
BinaryInputArchive bia = BinaryInputArchive.getArchive(new ByteBufferInputStream(incomingBuffer));

// 参数准备
ConnectRequest connReq = new ConnectRequest();
connReq.deserialize(bia, "connect");
boolean readOnly = false;

readOnly = bia.readBool("readOnly");
cnxn.isOldClient = false;

// 超时时间 : 4000
int sessionTimeout = connReq.getTimeOut();
// 连接密码 , 不存在则为一个空数组
byte passwd[] = connReq.getPasswd();

// 最小最大会话超时时间
int minSessionTimeout = getMinSessionTimeout();
int maxSessionTimeout = getMaxSessionTimeout();

// 设置超时时间
cnxn.setSessionTimeout(sessionTimeout);
cnxn.disableRecv();

// 如果session 存在
long sessionId = connReq.getSessionId();
if (sessionId != 0) {

long clientSessionId = connReq.getSessionId();
// 先关闭再重新打开会话
serverCnxnFactory.closeSession(sessionId);
cnxn.setSessionId(sessionId);
reopenSession(cnxn, sessionId, passwd, sessionTimeout);
} else {
// 创建一个会话
createSession(cnxn, passwd, sessionTimeout);
}
}

核心二 : Request 请求的处理
这里有一个稍微有点绕的多线程处理 , 后面再详细介绍一下 ,先看下主要的调用流程

  • C- PrepRequestProcessor # pRequest : 由配置类发起的first 请求处理器
  • C- SyncRequestProcessor # processRequest : 将请求加入 Queue 中(核心)
  • C- SyncRequestProcessor # run : 其中会一直循环处理 Request
  • C- FinalRequestProcessor # processRequest : 发起 Process 调用

PS : 其中循环的处理很不错 , 值得深入学习一下

四 . Zookeeper 存储的数据结构

另外一大重点就是了解一下数据是以什么样的结构保存到Zookeeper 中的 , Zookeeper 中存在以下几个核心的数据存储对象 :

  • ZKDatabase : 数据中心
  • DataTree : 数据数 , 核心数据对象

4.1 数据的获取

以数据的获取为例 , 经历了以下流程 :

  • C- SyncRequestProcessor # run : 注意 , 这个是一个不断从 queue 中获取数据的过程
  • C- FinalRequestProcessor # processRequest
  • C- ZKDatabase # getNode
  • C- DataTree # getNode
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
java复制代码public class DataTree {
// 核心存储对象
private final ConcurrentHashMap<String, DataNode> nodes = new ConcurrentHashMap<String, DataNode>();

// watches 对象集合
private final WatchManager dataWatches = new WatchManager();
private final WatchManager childWatches = new WatchManager();

/** the root of zookeeper tree */
private static final String rootZookeeper = "/";

/** the zookeeper nodes that acts as the management and status node **/
private static final String procZookeeper = Quotas.procZookeeper;

/** this will be the string thats stored as a child of root */
private static final String procChildZookeeper = procZookeeper.substring(1);

/**
* the zookeeper quota node that acts as the quota management node for
* zookeeper
*/
private static final String quotaZookeeper = Quotas.quotaZookeeper;

/** this will be the string thats stored as a child of /zookeeper */
private static final String quotaChildZookeeper = quotaZookeeper
.substring(procZookeeper.length() + 1);

/**
* the path trie that keeps track fo the quota nodes in this datatree
*/
private final PathTrie pTrie = new PathTrie();

}

总结

内容不多 , 但是比较重要 , 这几个环节弄清楚后 , 后面围着整个环节抽丝剥茧就行了

后续文章已经整理得差不多了 , 稍微修改一下后续发出来 , 文章写的比较早 , 版本比较旧 ,但是核心是差不多得

附录 :

Zookeeper 项目结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
java复制代码//─zookeeper-assembly                
//─zookeeper-client
│ └─zookeeper-client-c
//─zookeeper-compatibility-tests
│ └─zookeeper-compatibility-tests-curator
//─zookeeper-contrib
│ ├─zookeeper-contrib-fatjar
│ ├─zookeeper-contrib-huebrowser
│ ├─zookeeper-contrib-loggraph
│ ├─zookeeper-contrib-monitoring
│ ├─zookeeper-contrib-rest
│ ├─zookeeper-contrib-zkfuse
│ ├─zookeeper-contrib-zkperl
│ ├─zookeeper-contrib-zkpython
│ ├─zookeeper-contrib-zktreeutil
│ └─zookeeper-contrib-zooinspector
//─zookeeper-docs
//─zookeeper-it
//─zookeeper-jute
//─zookeeper-metrics-providers
│ └─zookeeper-prometheus-metrics
//─zookeeper-recipes
│ ├─zookeeper-recipes-election
│ ├─zookeeper-recipes-lock
│ └─zookeeper-recipes-queue
//─zookeeper-server

参考文档

www.cnblogs.com/jing99/p/12…

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%