一. 前言

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

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

二 . 源码的运行

2.1 主启动流程

// 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 , 大概了解了一下

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=2000
initLimit=10
syncLimit=5
dataDir=D:\\java\\workspace\\git\\zookeeper\\temp
clientPort=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<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 项目结构

//─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
来源：掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。