一. 前言
出于对集群选举流程的好奇 , 以是把 Zookeeper 源码拉下来跑了一下 , 这篇文档对这个过程做一个简单的记载.
要想看懂任何源码 , 第一步就是要跑起来 . 这一篇重要介绍 , 假如快速的跑源码 ,同时简单介绍其中的关键点 , 便于处理
二 . 源码的运行
2.1 主启动流程
// Step 1 : Git 拉取https://github.com/apache/zookeeper.git// Step 2 : 本地运行 (Intellij)1. 找到 zookeeper-server 子模块2. 找到 对应的启动类 3. 通过命令启动项目// PS : 这里查看了 Zookeeper 运行包 (ZkServer.cmd)的内容 , 决定先采用相同的方式启动 >>>>setlocalcall "%~dp0zkEnv.cmd"set ZOOMAIN=org.apache.zookeeper.server.quorum.QuorumPeerMainecho oncall %JAVA% "-Dzookeeper.log.dir=%ZOO_LOG_DIR%" "-Dzookeeper.root.logger=%ZOO_LOG4J_PROP%" -cp "%CLASSPATH%" %ZOOMAIN% "%ZOOCFG%" %*endlocal复制代码2.2 其他启动类
进入源码可以看到很多其他的启动类 , 这里查阅了一下 API Doc , 大概了解了一下
2.3 源码的重要逻辑
源码的重要入口类为 QuorumPeerMain , 而其他的类会由 QuorumPeerMain 发起调用 (比方 : ) , 这里提供一个简单的调用流程图 >>>
二 . 启动流程
来简单看一下 QuorumPeerMain 启动时做了什么 :
3.1 QuorumPeerMain 简介
当使用该类的main()方法启动程序时，第一个参数被用作配置文件的路径 , 配置文件中可以包含如下信息 :
- dataDir : ZooKeeper数据所在目录。
- dataLogDir : ZooKeeper事件日志存放目录。
- clientPort : 用于与客户端通讯的端口。
- tickTime : 一个滴答的持续时间，单位为毫秒。这是ZooKeeper中的基本时间单位。
- initLimit : 跟踪者等待与leader初始同步的最大节拍数。
- syncLimit : 跟踪者等待来自leader的消息(包罗心跳)的最大节拍数。
- server.id : 这是具有给定id的服务器将用于仲裁协议的主机:port[:port]。
# 以下是我的配置tickTime=2000initLimit=10syncLimit=5dataDir=D:\\java\\workspace\\git\\zookeeper\\tempclientPort=2181复制代码3.2 IDEA 配置
VM Options
"-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
D:\java\workspace\git\zookeeper\zoo.cfg复制代码
其中比较重要的就是 log 级别和你的源码路径 , 配置完成后一般项目就能正常跑起来了
三 . 请求与吸收
请求和吸收跑通了才是一切的基础 , Zk 的核心对象分别为 ClientCnxn 和 ServerCnxn
从 log 中不难发现 , 默认是走的 NIO
[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 : 发起处理请求
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
创建连接是一切的出发点 , 重要通过以下流程调用到该类
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
public class DataTree { // 核心存储对象 private final ConcurrentHashMap nodes = new ConcurrentHashMap(); // 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 项目结构
//─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复制代码参考文档
作者：AntBlack
链接：https://juejin.cn/post/7004833249566392351
来源：掘金
著作权归作者所有。商业转载请接洽作者得到授权，非商业转载请注明出处。
|
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
显身卡
