Zookeeper3.7源碼剖析

能力目標

• 能基於Maven導入最新版Zookeeper源碼
• 能說出Zookeeper單機啟動流程
• 理解Zookeeper預設通信中4個線程的作用
• 掌握Zookeeper業務處理源碼處理流程
• 能夠在Zookeeper源碼中Debug測試通信過程

1 Zookeeper源碼導入

Zookeeper是一個高可用的分散式數據管理和協調框架，並且能夠很好的保證分散式環境中數據的一致性。在越來越多的分散式系。在越來越多的分散式系統（Hadoop、HBase、Kafka）中，Zookeeper都作為核心組件使用。

我們當前課程主要是研究Zookeeper源碼，需要將Zookeeper工程導入到IDEA中，老版的zk是通過ant進行編譯的，但最新的zk(3.7)源碼中已經沒了build.xml，而多了pom.xml，也就是說構建方式由原先的Ant變成了Maven，源碼下下來後，直接編譯、運行是跑不起來的，有一些配置需要調整。

1.1 工程導入

Zookeeper各個版本源碼下載地址https://github.com/apache/zookeeper，我們可以在該倉庫下選擇不同的版本，我們選擇最新版本，當前最新版本為3.7，如下圖：

找到項目下載地址，我們選擇https地址，並複製該地址，通過該地址把項目導入到IDEA中。

點擊IDEA的VSC>Checkout from Version Controller>GitHub，操作如下圖：

克隆項目到本地：

項目導入本地後，效果如下：

項目運行的時候，缺一個版本對象，創建org.apache.zookeeper.version.Info，代碼如下：

public interface Info {
    public static final int MAJOR=3;
    public static final int MINOR=4;
    public static final int MICRO=6;
    public static final String QUALIFIER=null;
    public static final int REVISION=-1;
    public static final String REVISION_HASH = "1";
    public static final String BUILD_DATE="2020-12-03 09:29:06";
}

1.2 Zookeeper源碼錯誤解決

在zookeeper-server中找到org.apache.zookeeper.server.quorum.QuorumPeerMain並啟動該類，啟動前做如下配置：

啟動的時候會會報很多錯誤，比如缺包、缺對象，如下幾幅圖：

為瞭解決上面的錯誤，我們需要手動引入一些包，pom.xml引入如下依賴：

<!--引入依賴-->
<dependency>
  <groupId>io.dropwizard.metrics</groupId>
  <artifactId>metrics-core</artifactId>
  <version>3.1.0</version>
</dependency>
<dependency>
  <groupId>org.xerial.snappy</groupId>
  <artifactId>snappy-java</artifactId>
  <version>1.1.7.3</version>
</dependency>
<dependency>
  <groupId>org.eclipse.jetty</groupId>
  <artifactId>jetty-server</artifactId>
</dependency>
<dependency>
  <groupId>org.eclipse.jetty</groupId>
  <artifactId>jetty-servlet</artifactId>
</dependency>
<dependency>
  <groupId>commons-cli</groupId>
  <artifactId>commons-cli</artifactId>
</dependency>

1.3 Zookeeper命令(自學)

我們要想學習Zookeeper，需要先學會使用Zookeeper，它有很多豐富的命令，藉助這些命令可以深入理解Zookeeper，我們啟動源碼中的客戶端就可以使用Zookeeper相關命令。

啟動客戶端org.apache.zookeeper.ZooKeeperMain，如下圖：

啟動後，日誌如下：

1)節點列表:ls /

ls /
[dubbo, zookeeper]
ls /dubbo
[com.itheima.service.CarService]

2)查看節點狀態:stat /dubbo

stat /dubbo
cZxid = 0x3
ctime = Thu Dec 03 09:19:29 CST 2020
mZxid = 0x3
mtime = Thu Dec 03 09:19:29 CST 2020
pZxid = 0x4
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 13
numChildren = 1

節點信息參數說明如下：

3)創建節點:create /dubbo/code java

create /dubbo/code java
Created /dubbo/code

其中code表示節點，java表示節點下的內容。

4)查看節點數據:get /dubbo/code

get /dubbo/code
java

5)刪除節點:delete /dubbo/code || deleteall /dubbo/code

刪除沒有子節點的節點：delete /dubbo/code

刪除所有子節點：deleteall /dubbo/code

6)歷史操作命令:history

history
1 - ls /dubbo
2 - ls /dubbo/code
3 - get /dubbo/code
4 - get /dubbo/code
5 - create /dubbo/code java
6 - get /dubbo/code
7 - get /dubbo/code
8 - delete /dubbo/code
9 - get /dubbo/code
10 - listquota path
11 - history

1.4 Zookeeper分析工具

Zookeeper安裝比較方便，在安裝一個集群以後，查看數據卻比較麻煩，下麵介紹Zookeeper的數據查看工具——ZooInspector。

下載地址：https://issues.apache.org/jira/secure/attachment/12436620/ZooInspector.zip

下載壓縮包後，解壓後，我們需要運行zookeeper-dev-ZooInspector.jar：

輸入賬號密碼，就可以連接Zookeeper了，如下圖：

連接後，Zookeeper信息如下：

節點操作：增加節點、修改節點、刪除節點

1.5 Zookeeper案例應用

我們將資料中工程\dubbo工程導入到IDEA中，上圖是他們的調用關係，那麼問題來了：

• 生產者向Zookeeper註冊服務信息，Zookeeper把數據存哪兒了？
• 集群環境下，如果某個節點數據變更了，Zookeeper如何監聽到的？
• 集群環境下各個節點的數據如何同步？
• 如果某個節點掛了，Zookeeper如何選舉呢？
• ........

帶著上面的疑問，我們開始研究Zookeeper源碼。

2 ZK服務啟動流程源碼剖析

ZooKeeper可以以standalone、分散式的方式部署，standalone模式下只有一臺機器作為伺服器,ZooKeeper會喪失高可用特性，分散式是使用多個機器,每台機器上部署一個ZooKeeper伺服器,即使有伺服器宕機,只要少於半數,ZooKeeper集群依然可以正常對外提供服務，集群狀態下Zookeeper是具備高可用特性。

我們接下來對ZooKeeper以standalone模式啟動以及集群模式做一下源碼分析。

2.1 ZK單機/集群啟動流程

如上圖，上圖是Zookeeper單機/集群啟動流程，每個細節所做的事情都在上圖有說明，我們接下來按照流程圖對源碼進行分析。

2.2 ZK啟動入口分析

啟動入口類：QuorumPeerMain

該類是zookeeper單機/集群的啟動入口類，是用來載入配置、啟動QuorumPeer(選舉相關)線程、創建ServerCnxnFactory等，我們可以把代碼切換到該類的主方法(main)中，從該類的主方法開始分析，main方法代碼分析如下：

上面main方法雖然只是做了初始化配置，但調用了initializeAndRun()方法，initializeAndRun()方法中會根據配置來決定啟動單機Zookeeper還是集群Zookeeper，源碼如下：

如果啟動單機版，會調用ZooKeeperServerMain.main(args);，如果啟動集群版，會調用QuorumPeerMain.runFromConfig(config);，我們接下來對單機版啟動做源碼詳細剖析，集群版在後面章節中講解選舉機制時詳細講解。

2.3 ZK單機啟動源碼剖析

針對ZK單機啟動源碼方法調用鏈，我們已經提前做了一個方法調用關係圖，我們講解ZK單機啟動源碼，將和該圖進行一一匹對，如下圖：

1)單機啟動入口

按照上面的源碼分析，我們找到ZooKeeperServerMain.main(args)方法，該方法調用了ZooKeeperServerMain的initializeAndRun方法，在initializeAndRun方法中執行初始化操作，並運行Zookeeper服務，main方法如下：

2)配置文件解析

initializeAndRun()方法會註冊JMX，同時解析zoo.cfg配置文件，並調用runFromConfig()方法啟動Zookeeper服務，源碼如下：

3)單機啟動主流程

runFromConfig方法是單機版啟動的主要方法，該方法會做如下幾件事：

1:初始化各類運行指標，比如一次提交數據最大花費多長時間、批量同步數據大小等。
2:初始化許可權操作，例如IP許可權、Digest許可權。
3:創建事務日誌操作對象，Zookeeper中每次增加節點、修改數據、刪除數據都是一次事務操作，都會記錄日誌。
4:定義Jvm監控變數和常量，例如警告時間、告警閥值次數、提示閥值次數等。
5:創建ZookeeperServer，這裡只是創建，並不在ZooKeeperServerMain類中啟動。
6:啟動Zookeeper的控制台管理對象AdminServer，該對象採用Jetty啟動。
7:創建ServerCnxnFactory，該對象其實是Zookeeper網路通信對象，預設使用了NIOServerCnxnFactory。
8:在ServerCnxnFactory中啟動ZookeeperServer服務。
9:創建並啟動ContainerManager，該對象通過Timer定時執行,清理過期的容器節點和TTL節點,執行周期為分鐘。
10:防止主線程結束，阻塞主線程。

方法源碼如下：

4)網路通信對象創建

上面方法在創建網路通信對象的時候調用了ServerCnxnFactory.createFactory(),該方法其實是根據系統配置創建Zookeeper通信組件，可選的有NIOServerCnxnFactory(預設)和NettyServerCnxnFactory，關於通信對象我們會在後面進行詳細講解，該方法源碼如下：

5)單機啟動

cnxnFactory.startup(zkServer);方法其實就是啟動了ZookeeperServer，它調用NIOServerCnxnFactory的startup方法，該方法中會調用ZookeeperServer的startup方法啟動服務，ZooKeeperServerMain運行到shutdownLatch.await();主線程會阻塞住,源碼如下：

啟動後，日誌如下：

3 ZK網路通信源碼剖析

Zookeeper作為一個伺服器,自然要與客戶端進行網路通信,如何高效的與客戶端進行通信,讓網路IO不成為ZooKeeper的瓶頸是ZooKeeper急需解決的問題,ZooKeeper中使用ServerCnxnFactory管理與客戶端的連接,其有兩個實現,一個是NIOServerCnxnFactory,使用Java原生NIO實現;一個是NettyServerCnxnFactory,使用netty實現;使用ServerCnxn代表一個客戶端與服務端的連接。

從單機版啟動中可以發現Zookeeper預設通信組件為NIOServerCnxnFactory，他們和ServerCnxnFactory的關係如下圖：

3.1 NIOServerCnxnFactory工作流程

一般使用Java NIO的思路為使用1個線程組監聽OP_ACCEPT事件,負責處理客戶端的連接;使用1個線程組監聽客戶端連接的OP_READ和OP_WRITE事件,處理IO事件(netty也是這種實現方式).
但ZooKeeper並不是如此劃分線程功能的,NIOServerCnxnFactory啟動時會啟動四類線程：

1:accept thread:該線程接收來自客戶端的連接,並將其分配給selector thread(啟動一個線程)。

2:selector thread:該線程執行select(),由於在處理大量連接時,select()會成為性能瓶頸,因此啟動多個selector thread,使用系統屬性zookeeper.nio.numSelectorThreads配置該類線程數,預設個數為 核心數/2。

3:worker thread:該線程執行基本的套接字讀寫,使用系統屬性zookeeper.nio.numWorkerThreads配置該類線程數,預設為核心數∗2核心數∗2.如果該類線程數為0,則另外啟動一線程進行IO處理,見下文worker thread介紹。

4:connection expiration thread:若連接上的session已過期,則關閉該連接。

這四個線程在NIOServerCnxnFactory類上有說明，如下圖：

ZooKeeper中對線程需要處理的工作做了更細的拆分,解決了有大量客戶端連接的情況下,selector.select()會成為性能瓶頸,將selector.select()拆分出來,交由selector thread處理。

3.2 NIOServerCnxnFactory源碼

NIOServerCnxnFactory的源碼分析我們將按照上面所介紹的4個線程實現相關分析，並實現數據操作，在程式中獲取指定數據。

3.2.1 AcceptThread剖析

為了讓大家更容易理解AcceptThread，我們把它的結構和方法調用關係畫了一個詳細的流程圖，如下圖：

在NIOServerCnxnFactory類中有一個AccpetThread線程，為什麼說它是一個線程？我們看下它的繼承關係：AcceptThread > AbstractSelectThread > ZooKeeperThread > Thread，該線程接收來自客戶端的連接,並將其分配給selector thread(啟動一個線程)。

該線程執行流程：run執行selector.select(),並調用doAccept()接收客戶端連接，因此我們可以著重關註doAccept()方法,該類源碼如下：

doAccept()方法用於處理客戶端鏈接，當客戶端鏈接Zookeeper的時候，首先會調用該方法，調用該方法執行過程如下：

1:和當前服務建立鏈接。
2:獲取遠程客戶端電腦地址信息。
3:判斷當前鏈接是否超出最大限制。
4:調整為非阻塞模式。
5:輪詢獲取一個SelectorThread，將當前鏈接分配給該SelectorThread。
6:將當前請求添加到該SelectorThread的acceptedQueue中，並喚醒該SelectorThread。

doAccept()方法源碼如下：

上面代碼中addAcceptedConnection方法如下：

我們把項目中的分散式案例服務啟動，可以看到如下日誌列印：

AcceptThread----------鏈接服務的IP：127.0.0.1

3.2.2 SelectorThread剖析

同樣為了更容易梳理SelectorThread，我們也把它的結構和方法調用關係梳理成了流程圖，如下圖：

該線程的主要作用是從Socket讀取數據，並封裝成workRequest，並將workRequest交給workerPool工作線程池處理，同時將acceptedQueue中未處理的鏈接取出，並未每個鏈接綁定OP_READ讀事件，並封裝對應的上下文對象NIOServerCnxn。SelectorThread的run方法如下：

run()方法中會調用select()，而select()中的核心調用地方是handleIO()，我們看名字其實就知道這裡是處理客戶端請求的數據，但客戶端請求數據並非在SelectorThread線程中處理，我們接著看handleIO()方法。

handleIO()方法會封裝當前SelectorThread為IOWorkRequest,並將IOWorkRequest交給workerPool來調度，而workerPool調度才是讀數據的開始，源碼如下：

3.2.3 WorkerThread剖析

WorkerThread相比上面的線程而言，調用關係頗為複雜，設計到了多個對象方法調用，主要用於處理IO，但並未對數據做出處理，數據處理將有業務鏈對象RequestProcessor處理，調用關係圖如下：

ZooKeeper中通過WorkerService管理一組worker thread線程,前面我們在看SelectorThread的時候，能夠看到workerPool的schedule方法被執行，如下圖：

我們跟蹤workerPool.schedule(workRequest);可以發現它調用了WorkerService.schedule(workRequest) > WorkerService.schedule(WorkRequest, long)，該方法創建了一個新的線程ScheduledWorkRequest,並啟動了該線程，源碼如下：

ScheduledWorkRequest實現了Runnable介面，併在run()方法中調用了IOWorkRequest中的doWork方法，在該方法中會調用doIO執行IO數據處理，源碼如下：

IOWorkRequest的doWork源碼如下：

接下來的調用鏈路比較複雜，我們把核心步驟列出，在能直接看到數據讀取的地方詳細分析源碼。上面方法調用鏈路：NIOServerCnxn.doIO()>readPayload()>readRequest() >ZookeeperServer.processPacket() ，最後一步方法是獲取核心數據的地方，我們可以修改下代碼讀取數據：

添加測試代碼如下：

//==========測試 Start===========
//定義接收輸入流對象（輸出流）
ByteArrayOutputStream os = new ByteArrayOutputStream();

//將網路輸入流讀取到輸出流中
byte[] buffer = new byte[1024];
int len=0;
while ((len=bais.read(buffer))!=-1){
    os.write(buffer,0,len);
}
String result = new String(os.toByteArray(),"UTF-8");
System.out.println("processPacket---------------讀到的數據："+result);
//==========測試 End===========

我們啟動客戶端創建一個demo節點，並添加數據為 abcdefg

create /demo abcdefg

控制台數據如下：

測試完成後，不要忘了將該測試註釋掉。我們可以執行其他增刪改查操作，可以輸出RequestHeader.type查看操作類型，操作類型代碼在ZooDefs中有標識,常用的操作類型如下：

int create = 1;
int delete = 2;
int exists = 3;
int getData = 4;
int setData = 5;
int getACL = 6;
int setACL = 7;
int getChildren = 8;
int sync = 9;
int ping = 11;

2.3.4 ConnectionExpirerThread剖析

後臺啟動ConnectionExpirerThread清理線程清理過期的session,線程中無限迴圈,執行工作如下:

2.3 ZK通信優劣總結

Zookeeper在通信方面預設使用了NIO，並支持擴展Netty實現網路數據傳輸。相比傳統IO，NIO在網路數據傳輸方面有很多明顯優勢：

1:傳統IO在處理數據傳輸請求時，針對每個傳輸請求生成一個線程，如果IO異常，那麼線程阻塞，在IO恢復後喚醒處理線程。在同時處理大量連接時，會實例化大量的線程對象。每個線程的實例化和回收都需要消耗資源，jvm需要為其分配TLAB，然後初始化TLAB，最後綁定線程，線程結束時又需要回收TLAB，這些都需要CPU資源。

2:NIO使用selector來輪詢IO流，內部使用poll或者epoll，以事件驅動形式來相應IO事件的處理。同一時間只需實例化很少的線程對象，通過對線程的復用來提高CPU資源的使用效率。

3:CPU輪流為每個線程分配時間片的形式，間接的實現單物理核處理多線程。當線程越多時，每個線程分配到的時間片越短，或者迴圈分配的周期越長，CPU很多時間都耗費在了線程的切換上。線程切換包含線程上個線程數據的同步(TLAB同步)，同步變數同步至主存，下個線程數據的載入等等，他們都是很耗費CPU資源的。

4:在同時處理大量連接，但活躍連接不多時，NIO的事件響應模式相比於傳統IO有著極大的性能提升。NIO還提供了FileChannel，以zero-copy的形式傳輸數據，相較於傳統的IO，數據不需要拷貝至用戶空間，可直接由物理硬體(磁碟等)通過內核緩衝區後直接傳遞至網關，極大的提高了性能。

5:NIO提供了MappedByteBuffer，其將文件直接映射到記憶體（這裡的記憶體指的是虛擬記憶體，並不是物理記憶體），能極大的提高IO吞吐能力。

ZK在使用NIO通信雖然大幅提升了數據傳輸能力，但也存在一些代碼詬病問題：

1:Zookeeper通信源碼部分學習成本高，需要掌握NIO和多線程
2:多線程使用頻率高，消耗資源多，但性能得到提升
3:Zookeeper數據處理調用鏈路複雜，多處存在內部類，代碼結構不清晰，寫法比較經典

4 RequestProcessor處理請求源碼剖析

zookeeper 的業務處理流程就像工作流一樣，其實就是一個單鏈表；在zookeeper啟動的時候，會確立各個節點的角色特性，即leader、follower和observer，每個角色確立後，就會初始化它的工作責任鏈；

4.1 RequestProcessor結構

客戶端請求過來，每次執行不同事務操作的時候，Zookeeper也提供了一套業務處理流程RequestProcessor，RequestProcessor的處理流程如下圖：

我們來看一下RequestProcessor初始化流程，ZooKeeperServer.setupRequestProcessors()方法源碼如下：

它的創建步驟：

1:創建finalProcessor。
2:創建syncProcessor，並將finalProcessor作為它的下一個業務鏈。
3:啟動syncProcessor。
4:創建firstProcessor(PrepRequestProcessor)，將syncProcessor作為firstProcessor的下一個業務鏈。
5:啟動firstProcessor。

syncProcessor創建時，將finalProcessor作為參數傳遞進來源碼如下：

firstProcessor創建時，將syncProcessor作為參數傳遞進來源碼如下：

PrepRequestProcessor/SyncRequestProcessor關係圖：

PrepRequestProcessor和SyncRequestProcessor的結構一樣，都是實現了Thread的一個線程，所以在這裡初始化時便啟動了這兩個線程。

4.2 PrepRequestProcessor剖析

PrepRequestProcessor是請求處理器的第1個處理器，我們把之前的請求業務處理銜接起來，一步一步分析。ZooKeeperServer.processPacket()>submitRequest()>enqueueRequest()>RequestThrottler.submitRequest() ，我們來看下RequestThrottler.submitRequest()源碼,它將當前請求添加到submittedRequests隊列中了，源碼如下：

而RequestThrottler繼承了 ZooKeeperCriticalThread > ZooKeeperThread > Thread，也就是說當前RequestThrottler是個線程，我們看看它的run方法做了什麼事，源碼如下：

RequestThrottler調用了ZooKeeperServer.submitRequestNow()方法，而該方法又調用了firstProcessor的方法，源碼如下：

ZooKeeperServer.submitRequestNow()方法調用了firstProcessor.processRequest()方法，而這裡的firstProcessor就是初始化業務處理鏈中的PrepRequestProcessor，也就是說三個RequestProecessor中最先調用的是PrepRequestProcessor。

PrepRequestProcessor.processRequest()方法將當前請求添加到了隊列submittedRequests中，源碼如下：

上面方法中並未從submittedRequests隊列中獲取請求，如何執行請求的呢，因為PrepRequestProcessor是一個線程，因此會在run中執行，我們查看run方法源碼的時候發現它調用了pRequest()方法，pRequest()方法源碼如下：

首先先執行pRequestHelper()方法，該方法是PrepRequestProcessor處理核心業務流程，主要是一些過濾操作，操作完成後，會將請求交給下一個業務鏈，也就是SyncRequestProcessor.processRequest()方法處理請求。

我們來看一下PrepRequestProcessor.pRequestHelper()方法做了哪些事，源碼如下：

從上面源碼可以看出PrepRequestProcessor.pRequestHelper()方法判斷了客戶端操作類型，但無論哪種操作類型幾乎都調用了pRequest2Txn()方法，我們來看看源碼：

從上面代碼可以看出pRequest2Txn()方法主要做了許可權校驗、快照記錄、事務信息記錄相關的事，還並未涉及數據處理，也就是說PrepRequestProcessor其實是做了操作前許可權校驗、快照記錄、事務信息記錄相關的事。

我們DEBUG調試一次，看看業務處理流程是否和我們上面所分析的一致。

添加節點：

create /zkdemo itheima

DEBUG測試如下：

客戶端請求先經過ZooKeeperServer.submitRequestNow()方法，並調用firstProcessor.processRequest()方法，而firstProcessor=PrepRequestProcessor，如下圖：

進入PrepRequestProcessor.pRequest()方法，執行完pRequestHelper()方法後，開始執行下一個業務鏈的方法，而下一個業務鏈nextProcessor=SyncRequestProcessor，如下測試圖：

4.3 SyncRequestProcessor剖析

分析了PrepRequestProcessor處理器後，接著來分析SyncRequestProcessor，該處理器主要是將請求數據高效率存入磁碟，並且請求在寫入磁碟之前是不會被轉發到下個處理器的。

我們先看請求被添加到隊列的方法：

同樣SyncRequestProcessor是一個線程，執行隊列中的請求也線上程中觸發，我們看它的run方法，源碼如下：

run方法會從queuedRequests隊列中獲取一個請求，如果獲取不到就會阻塞等待直到獲取到一個請求對象，程式才會繼續往下執行，接下來會調用Snapshot Thread線程實現將客戶端發送的數據以快照的方式寫入磁碟，最終調用flush()方法實現數據提交，flush()方法源碼如下：

flush()方法實現了數據提交，並且會將請求交給下一個業務鏈，下一個業務鏈為FinalRequestProcessor。

4.4 FinalRequestProcessor剖析

前面分析了SyncReqeustProcessor，接著分析請求處理鏈中最後的一個處理器FinalRequestProcessor,該業務處理對象主要用於返回Response。

4.5 ZK業務鏈處理優劣總結

Zookeeper業務鏈處理，思想遵循了AOP思想，但並未採用相關技術，為了提升效率，仍然大幅使用到了多線程。正因為有了業務鏈路處理先後順序，使得Zookeeper業務處理流程更清晰更容易理解，但大量混入了多線程，也似的學習成本增加。

本文由傳智教育博學谷 - 狂野架構師教研團隊發佈
如果本文對您有幫助，歡迎關註和點贊；如果您有任何建議也可留言評論或私信，您的支持是我堅持創作的動力
轉載請註明出處！