聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境
一、前言
老周這里編譯 Kafka 的版本是 2.7,為啥采用這個(gè)版本來搭建源碼的閱讀環(huán)境呢?因?yàn)樵摪姹鞠鄬?duì)來說比較新。而我為啥不用 2.7 后的版本呢?比如 2.8,這是因?yàn)槿サ袅?ZooKeeper,還不太穩(wěn)定,生產(chǎn)環(huán)境也不太建議使用,所以以 2.7 版本進(jìn)行源碼搭建并研究。二、環(huán)境準(zhǔn)備
- JDK:1.8.0_241
- Scala:2.12.8
- Gradle:6.6
- Zookeeper:3.4.14
三、環(huán)境搭建
3.1 JDK 環(huán)境搭建這個(gè)就不用我說了吧,搞 Java 的本機(jī)都有 JDK 環(huán)境。3.2 Scala 環(huán)境搭建下載鏈接:https://www.scala-lang.org/download/2.12.8.html
vim?~/.bash_profile
#?這里的路徑是你安裝
SCALA_HOME=/Users/Riemann/Tools/scala-2.12.8
export?SCALA_HOME
export?PATH=$PATH:$SCALA_HOME/bin
#?使環(huán)境變量生效,在命令行執(zhí)行。
source??~/.bash_profile
3.2.2 驗(yàn)證終端輸入以下命令:scala?-version
出現(xiàn)以下提示,說明 Scala 環(huán)境搭建成功。

我們選擇想要安裝的發(fā)布版本,gradle-x.x-bin.zip 是需要下載的安裝發(fā)布版,gradle-x.x-src.zip 是源碼,gradle-x.x-all.zip 則是下載全部的文件。我本地為 gradle-6.6。Gradle下載的源碼不需要安裝,我們將下載的壓縮包在本機(jī)的目錄下直接解壓即可,解壓后的目錄如下圖所示。

vim?~/.bash_profile
#?這里的路徑是你安裝
GRADLE_HOME=/Users/Riemann/Tools/gradle-6.6
export?GRADLE_HOME
export?PATH=$PATH:$GRADLE_HOME/bin
#?使環(huán)境變量生效,在命令行執(zhí)行。
source??~/.bash_profile
3.3.2 驗(yàn)證終端輸入以下命令:gradle?-v
出現(xiàn)以下提示,說明 Gradle 環(huán)境搭建成功。
3.4 Zookeeper 環(huán)境搭建Zookeeper 環(huán)境老周在 Linux 環(huán)境已經(jīng)搭建好了的,直接用。這里我也給出搭建的步驟,不管你是啥系統(tǒng),都是類似的~3.4.1 下載
wget?http://mirrors.hust.edu.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz
3.4.2 解壓tar?-zxvf?zookeeper-3.4.14.tar.gz
3.4.3 進(jìn)入 zookeeper-3.4.14 目錄,創(chuàng)建 data 文件夾?cd?zookeeper-3.4.14?
?mkdir?data
3.4.4 修改配置文件cd?conf
mv?zoo_sample.cfg?zoo.cfg
3.4.5 修改 zoo.cfg 中的 data 屬性dataDir=/root/zookeeper-3.4.14/data
3.4.6 zookeeper 服務(wù)啟動(dòng)進(jìn)入 bin 目錄,啟動(dòng)服務(wù)輸入命令./zkServer.sh?start
輸出以下內(nèi)容表示啟動(dòng)成功

下載完后解壓,這個(gè)源碼文件還需要導(dǎo)入依賴 jar 包,個(gè)人使用 IDEA 來 import 導(dǎo)入項(xiàng)目,導(dǎo)入完后需使用前面配置好的 gradle 作為 Gradle home 地址。3.5.1 導(dǎo)入 Kafka 源碼至 IDEA



3.5.2 修改 build.gradle接下來還不能導(dǎo) jar 包,需要把鏡像文件下載服務(wù)器更換為國(guó)內(nèi)的私服,否則會(huì)相當(dāng)慢,直接導(dǎo)致 "time out" 報(bào)錯(cuò)。進(jìn)入 kafka 源碼包,修改 build.gradle 文件,在原來配置上,添加 ali 私服配置。
buildscript?{
????repositories?{
????????maven?{
????????????url?'http://maven.aliyun.com/nexus/content/groups/public/'
????????}
????????maven?{
????????????url?'http://maven.aliyun.com/nexus/content/repositories/jcenter'
????????}
????}
}
allprojects?{
????repositories?{
????????maven?{
????????????url?'http://maven.aliyun.com/nexus/content/groups/public/'
????????}
????????maven?{
????????????url?'http://maven.aliyun.com/nexus/content/repositories/jcenter'
????????}
????}
}

3.5.3 代碼構(gòu)建可以用命令來構(gòu)建,也可以在 idea 圖形界面的 gradle 來構(gòu)建,這里肯定是 idea 圖形化界面操作更簡(jiǎn)單,但這里也提供 gradle 的命令來構(gòu)建。
./gradlew?clean?build?-x?test
去找一下直接下載 Wrapper 所需的 Jar 包,手動(dòng)把這個(gè) Jar 文件拷貝到 kafka 路徑下的 gradle/wrapper 子目錄下,然后重新執(zhí)行 gradlew build 命令去構(gòu)建工程。鏈接: https://pan.baidu.com/s/1W6EHysWY3ZWQZRWNdNZn3Q 提取碼: hpj5gradle 其它命令:#?構(gòu)建?jar包并運(yùn)行
./gradlew?jar
#?構(gòu)建項(xiàng)目,看你是idea工具還是eclipse
./gradlew?idea
./gradlew?eclipse
#?構(gòu)建源碼包
./gradlew?srcJar
#?構(gòu)建javadoc文檔
./gradlew?aggregatedJavadoc
#?清理并構(gòu)建
./gradlew?clean
四、代碼結(jié)構(gòu)

- bin 目錄:保存 Kafka 工具行腳本,我們熟知的 kafka-server-start 和 kafka-console-producer 等腳本都存放在這里。checkstyle 目錄:代碼規(guī)范,自動(dòng)化檢測(cè)。
Checkstyle 是什么,關(guān)于格式化的討論就不曾中斷過,到底什么才是正確的,什么才是錯(cuò)誤的,到現(xiàn)在也沒有完整的定論。但隨著時(shí)間發(fā)展,漸漸衍生出一套規(guī)范出來。沒有什么絕對(duì)的正確和錯(cuò)誤,關(guān)鍵在于規(guī)范的定義。最出名的就是 google style guide,Checkstyle 就是以這種風(fēng)格開發(fā)出的一個(gè)自動(dòng)化插件,來輔助判斷代碼格式是否滿足規(guī)范。
該目錄下的文件定義了工程代碼格式的規(guī)范,我們可以在 build.gradle 中看到相關(guān) checkstyle 的配置和自動(dòng)化代碼格式化配置:checkstyle 配置:
scala 自動(dòng)化代碼格式化配置: - clients 目錄:保存 Kafka 客戶端代碼,比如生產(chǎn)者和消費(fèi)者的代碼都在該目錄下。
- config 目錄:保存 Kafka 的配置文件,其中比較重要的配置文件是 server.properties。
- connect 目錄:保存 Connect 組件的源代碼。Kafka Connect 組件是用來實(shí)現(xiàn) Kafka 與外部系統(tǒng)之間的實(shí)時(shí)數(shù)據(jù)傳輸?shù)摹?/p>
- core 目錄:保存 Broker 端代碼。Kafka 服務(wù)器端代碼全部保存在該目錄下。
- docs 目錄:Kafka 設(shè)計(jì)文檔以及組件相關(guān)結(jié)構(gòu)圖。
- examples 目錄:Kafka 樣例相關(guān)目錄。
- generator 目錄:Kafka 消息類處理模塊,主要是根據(jù) clients 模塊下的 message json 文件生成對(duì)應(yīng)的 java 類,在 build.gradle 文件中,可以看到定義了一個(gè)任務(wù) processMessages:
- gradle 目錄:gradle 的腳本和依賴包定義等相關(guān)文件。
- jmh-benchmarks 目錄:Kafka 代碼微基準(zhǔn)測(cè)試相關(guān)類。
JMH,即 Java Microbenchmark Harness,是專門用于代碼微基準(zhǔn)測(cè)試的工具套件。何謂 Micro Benchmark 呢?簡(jiǎn)單的來說就是基于方法層面的基準(zhǔn)測(cè)試,精度可以達(dá)到微秒級(jí)。當(dāng)你定位到熱點(diǎn)方法,希望進(jìn)一步優(yōu)化方法性能的時(shí)候,就可以使用 JMH 對(duì)優(yōu)化的結(jié)果進(jìn)行量化的分析。
JMH 比較典型的應(yīng)用場(chǎng)景有: - 想準(zhǔn)確的知道某個(gè)方法需要執(zhí)行多長(zhǎng)時(shí)間,以及執(zhí)行時(shí)間和輸入之間的相關(guān)性;
- 對(duì)比接口不同實(shí)現(xiàn)在給定條件下的吞吐量,找到最優(yōu)實(shí)現(xiàn)。
- kafka-logs 目錄:server.properties 文件中配置 log.dirs 生成的目錄。
- log4j-appender 目錄:
A log4j appender that produces log messages to Kafka
這個(gè)目錄里面就一個(gè) KafkaLog4jAppender 類。 - raft 目錄:raft 一致性協(xié)議相關(guān)。
- streams 目錄:
Kafka Streams is a client library for building applications and microservices, where the input and output data are stored in Kafka clusters.
提供一個(gè)基于 Kafka 的流式處理類庫,直接提供具體的類給開發(fā)者調(diào)用,整個(gè)應(yīng)用的運(yùn)行方式主要由開發(fā)者控制,方便使用和調(diào)試。Kafka Streams 是一個(gè)用來構(gòu)建流處理程序的庫,特別是其輸入是一個(gè) Kafka topic,輸出是另一個(gè) Kafka topic 的程序(或者是調(diào)用外部服務(wù),或者是更新數(shù)據(jù)庫,或者其它)。它使得你以一種分布式以及容錯(cuò)的方式來做這件事情。 - tests 目錄:此目錄的內(nèi)容介紹如何進(jìn)行 Kafka 系統(tǒng)集成和性能測(cè)試。
- tools 目錄:工具類模塊。
- vagrant 目錄:介紹如何在 Vagrant 虛擬環(huán)境中運(yùn)行 Kafka,提供了相關(guān)的腳本文件和說明文檔。
Vagrant 是一個(gè)基于 Ruby 的工具,用于創(chuàng)建和部署虛擬化開發(fā)環(huán)境。它使用 Oracle 的開源 VirtualBox 虛擬化系統(tǒng),使用 Chef 創(chuàng)建自動(dòng)化虛擬環(huán)境。

- admin 包:執(zhí)行管理命令的功能;
- api 包:封裝請(qǐng)求和響應(yīng) DTO 對(duì)象;
- cluster 包:集群對(duì)象,例如 Replica 類代表一個(gè)分區(qū)副本,Partition 類代表一個(gè)分區(qū);
- common 包:通用 jar 包;
- controller包:和kafkaController(kc)相關(guān)的類,重點(diǎn)模塊,一個(gè)kafka集群只有一個(gè)leader kc,該kc負(fù)責(zé) 分區(qū)管理,副本管理,并保證集群信息在集群中同步;
- coordinator 包:保存了消費(fèi)者端的 GroupCoordinator 代碼和用于事務(wù)的 TransactionCoordinator 代碼。對(duì) coordinator 包進(jìn)行分析,特別是對(duì)消費(fèi)者端的 GroupCoordinator 代碼進(jìn)行分析,是 Broker 端協(xié)調(diào)者組件設(shè)計(jì)原理的關(guān)鍵。
- log 包:保存了 Kafka 最核心的日志結(jié)構(gòu)代碼,包括日志、日志段、索引文件等, 另外,該包下還封裝了 Log Compaction 的實(shí)現(xiàn)機(jī)制,是非常重要的源碼包。
- network 包:封裝了 Kafka 服務(wù)器端網(wǎng)絡(luò)層的代碼,特別是 SocketServer.scala 這個(gè)文件,是 Kafka 實(shí)現(xiàn) Reactor 模式的具體操作類,非常值得一讀。
- consumer 包:后面會(huì)丟棄該包,用 clients 包下 consumer 相關(guān)類代替。
- server 包:顧名思義,它是 Kafka 的服務(wù)器端主代碼,里面的類非常多,很多關(guān)鍵的 Kafka 組件都存放在這里,比如狀態(tài)機(jī)、Purgatory 延時(shí)機(jī)制等。
- tools 包:工具類。
五、環(huán)境驗(yàn)證
下面我們來驗(yàn)證一下 Kafka 源碼環(huán)境是否搭建成功。5.1 首先,我們?cè)?core/src/main 目錄下新建 resources 目錄,再將 conf 目錄下的 log4j.properties 配置文件拷貝到 resources 目錄下。如下圖所示:
5.2 修改 conf 目錄下的 server.properties 文件
log.dirs=/Users/Riemann/Code/framework-source-code-analysis/kafka-2.7.0-src/kafka-logs
server.properties 文件中的其他配置暫時(shí)不用修改。5.3 在 IDEA 中配置 kafka.Kafka 這個(gè)入口類具體配置如下圖所示:
5.4 啟動(dòng) Kafka Broker啟動(dòng)成功的話,控制臺(tái)輸出沒有異常,且能看到如下輸出:

log4j:WARN?No?appenders?could?be?found?for?logger?(kafka.utils.Log4jControllerRegistration$).
log4j:WARN?Please?initialize?the?log4j?system?properly.
log4j:WARN?See?http://logging.apache.org/log4j/1.2/faq.html#noconfig?for?more?info.
在 project structure 中加入 slf4j-log4j12-1.7.30.jar 和 log4j-1.2.17.jar 兩個(gè)日志包,當(dāng)然也可以在 build.gradle 中添加對(duì)應(yīng)的配置來添加包。方法1:
方法2:
compile?group:?'log4j',?name:?'log4j',?version:?'1.2.17'
compile?group:?'org.slf4j',?name:?'slf4j-api',?version:?'1.7.30'
compile?group:?'org.slf4j',?name:?'slf4j-log4j12',?version:?'1.7.30'
加到 build.gradle 文件中的 core 模塊:
5.5.2 異常2
SLF4J:?Failed?to?load?class?"org.slf4j.impl.StaticLoggerBinder".
SLF4J:?Defaulting?to?no-operation?(NOP)?logger?implementation
SLF4J:?See?http://www.slf4j.org/codes.html#StaticLoggerBinder?for?further?details.

5.6 發(fā)送、消費(fèi) message我們這里使用 Kafka 自帶的腳本工具來驗(yàn)證上面搭建的 Kafka 源碼環(huán)境首先,我們進(jìn)入到 ${KAFKA_HOME}/bin 目錄,通過 kafka-topics.sh 命令來創(chuàng)建一個(gè)名為 topic_test 的 topic:執(zhí)行效果如下圖所示:

然后我們通過 kafka-console-consumer.sh 命令啟動(dòng)一個(gè)命令行的 consumer 來消費(fèi) topic_test 這個(gè) topic,如下:
./kafka-console-consumer.sh?--bootstrap-server?localhost:9092?--topic?topic_test

接下來,我們通過 kafka-console-producer.sh 命令啟動(dòng)一個(gè)命令行的 producer 向 topic_test 這個(gè) topic 中生成數(shù)據(jù),如下:

當(dāng)我們輸入一條 message 并回車之后,message 會(huì)發(fā)送到 topic_test 這個(gè) topic 中。


大功告成,后續(xù)會(huì)陸續(xù)分析 Kafka Broker 端的源碼,盡情期待~