[導讀]一、前言老周這里編譯Kafka的版本是2.7，為啥采用這個版本來搭建源碼的閱讀環(huán)境呢？因為該版本相對來說比較新。而我為啥不用2.7后的版本呢？比如2.8，這是因為去掉了ZooKeeper，還不太穩(wěn)定，生產(chǎn)環(huán)境也不太建議使用，所以以2.7版本進行源碼搭建并研究。二、環(huán)境準備JDK：...
一、前言
老周這里編譯 Kafka 的版本是 2.7，為啥采用這個版本來搭建源碼的閱讀環(huán)境呢？因為該版本相對來說比較新。而我為啥不用 2.7 后的版本呢？比如 2.8，這是因為去掉了 ZooKeeper，還不太穩(wěn)定，生產(chǎn)環(huán)境也不太建議使用，所以以 2.7 版本進行源碼搭建并研究。二、環(huán)境準備
- JDK：1.8.0_241
- Scala：2.12.8
- Gradle：6.6
- Zookeeper：3.4.14
三、環(huán)境搭建
3.1 JDK 環(huán)境搭建這個就不用我說了吧，搞 Java 的本機都有 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 驗證終端輸入以下命令：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下載的源碼不需要安裝，我們將下載的壓縮包在本機的目錄下直接解壓即可，解壓后的目錄如下圖所示。
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 驗證終端輸入以下命令：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 進入 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ù)啟動進入 bin 目錄，啟動服務(wù)輸入命令./zkServer.sh?start
輸出以下內(nèi)容表示啟動成功下載完后解壓，這個源碼文件還需要導入依賴 jar 包，個人使用 IDEA 來 import 導入項目，導入完后需使用前面配置好的 gradle 作為 Gradle home 地址。3.5.1 導入 Kafka 源碼至 IDEA
3.5.2 修改 build.gradle接下來還不能導 jar 包，需要把鏡像文件下載服務(wù)器更換為國內(nèi)的私服，否則會相當慢，直接導致 "time out" 報錯。進入 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 圖形化界面操作更簡單，但這里也提供 gradle 的命令來構(gòu)建。
./gradlew?clean?build?-x?test
去找一下直接下載 Wrapper 所需的 Jar 包，手動把這個 Jar 文件拷貝到 kafka 路徑下的 gradle/wrapper 子目錄下，然后重新執(zhí)行 gradlew build 命令去構(gòu)建工程。鏈接: https://pan.baidu.com/s/1W6EHysWY3ZWQZRWNdNZn3Q 提取碼: hpj5gradle 其它命令：#?構(gòu)建?jar包并運行
./gradlew?jar
#?構(gòu)建項目，看你是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ī)范，自動化檢測。
Checkstyle 是什么，關(guān)于格式化的討論就不曾中斷過，到底什么才是正確的，什么才是錯誤的，到現(xiàn)在也沒有完整的定論。但隨著時間發(fā)展，漸漸衍生出一套規(guī)范出來。沒有什么絕對的正確和錯誤，關(guān)鍵在于規(guī)范的定義。最出名的就是 google style guide，Checkstyle 就是以這種風格開發(fā)出的一個自動化插件，來輔助判斷代碼格式是否滿足規(guī)范。
該目錄下的文件定義了工程代碼格式的規(guī)范，我們可以在 build.gradle 中看到相關(guān) checkstyle 的配置和自動化代碼格式化配置：checkstyle 配置：
scala 自動化代碼格式化配置： - clients 目錄：保存 Kafka 客戶端代碼，比如生產(chǎn)者和消費者的代碼都在該目錄下。
- config 目錄：保存 Kafka 的配置文件，其中比較重要的配置文件是 server.properties。
- connect 目錄：保存 Connect 組件的源代碼。Kafka Connect 組件是用來實現(xiàn) Kafka 與外部系統(tǒng)之間的實時數(shù)據(jù)傳輸?shù)摹?/p>
- core 目錄：保存 Broker 端代碼。Kafka 服務(wù)器端代碼全部保存在該目錄下。
- docs 目錄：Kafka 設(shè)計文檔以及組件相關(guān)結(jié)構(gòu)圖。
- examples 目錄：Kafka 樣例相關(guān)目錄。
- generator 目錄：Kafka 消息類處理模塊，主要是根據(jù) clients 模塊下的 message json 文件生成對應(yīng)的 java 類，在 build.gradle 文件中，可以看到定義了一個任務(wù) processMessages：
- gradle 目錄：gradle 的腳本和依賴包定義等相關(guān)文件。
- jmh-benchmarks 目錄：Kafka 代碼微基準測試相關(guān)類。
JMH，即 Java Microbenchmark Harness，是專門用于代碼微基準測試的工具套件。何謂 Micro Benchmark 呢？簡單的來說就是基于方法層面的基準測試，精度可以達到微秒級。當你定位到熱點方法，希望進一步優(yōu)化方法性能的時候，就可以使用 JMH 對優(yōu)化的結(jié)果進行量化的分析。
JMH 比較典型的應(yīng)用場景有： - 想準確的知道某個方法需要執(zhí)行多長時間，以及執(zhí)行時間和輸入之間的相關(guān)性；
- 對比接口不同實現(xiàn)在給定條件下的吞吐量，找到最優(yōu)實現(xiàn)。
- kafka-logs 目錄：server.properties 文件中配置 log.dirs 生成的目錄。
- log4j-appender 目錄：
A log4j appender that produces log messages to Kafka
這個目錄里面就一個 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.
提供一個基于 Kafka 的流式處理類庫，直接提供具體的類給開發(fā)者調(diào)用，整個應(yīng)用的運行方式主要由開發(fā)者控制，方便使用和調(diào)試。Kafka Streams 是一個用來構(gòu)建流處理程序的庫，特別是其輸入是一個 Kafka topic，輸出是另一個 Kafka topic 的程序（或者是調(diào)用外部服務(wù)，或者是更新數(shù)據(jù)庫，或者其它）。它使得你以一種分布式以及容錯的方式來做這件事情。 - tests 目錄：此目錄的內(nèi)容介紹如何進行 Kafka 系統(tǒng)集成和性能測試。
- tools 目錄：工具類模塊。
- vagrant 目錄：介紹如何在 Vagrant 虛擬環(huán)境中運行 Kafka，提供了相關(guān)的腳本文件和說明文檔。
Vagrant 是一個基于 Ruby 的工具，用于創(chuàng)建和部署虛擬化開發(fā)環(huán)境。它使用 Oracle 的開源 VirtualBox 虛擬化系統(tǒng)，使用 Chef 創(chuàng)建自動化虛擬環(huán)境。
- admin 包：執(zhí)行管理命令的功能；
- api 包：封裝請求和響應(yīng) DTO 對象；
- cluster 包：集群對象，例如 Replica 類代表一個分區(qū)副本，Partition 類代表一個分區(qū)；
- common 包：通用 jar 包；
- controller包：和kafkaController（kc）相關(guān)的類，重點模塊，一個kafka集群只有一個leader kc，該kc負責 分區(qū)管理，副本管理，并保證集群信息在集群中同步；
- coordinator 包：保存了消費者端的 GroupCoordinator 代碼和用于事務(wù)的 TransactionCoordinator 代碼。對 coordinator 包進行分析，特別是對消費者端的 GroupCoordinator 代碼進行分析，是 Broker 端協(xié)調(diào)者組件設(shè)計原理的關(guān)鍵。
- log 包：保存了 Kafka 最核心的日志結(jié)構(gòu)代碼，包括日志、日志段、索引文件等, 另外，該包下還封裝了 Log Compaction 的實現(xiàn)機制，是非常重要的源碼包。
- network 包：封裝了 Kafka 服務(wù)器端網(wǎng)絡(luò)層的代碼，特別是 SocketServer.scala 這個文件，是 Kafka 實現(xiàn) Reactor 模式的具體操作類，非常值得一讀。
- consumer 包：后面會丟棄該包，用 clients 包下 consumer 相關(guān)類代替。
- server 包：顧名思義，它是 Kafka 的服務(wù)器端主代碼，里面的類非常多，很多關(guān)鍵的 Kafka 組件都存放在這里，比如狀態(tài)機、Purgatory 延時機制等。
- tools 包：工具類。
五、環(huán)境驗證
下面我們來驗證一下 Kafka 源碼環(huán)境是否搭建成功。5.1 首先，我們在 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 文件中的其他配置暫時不用修改。5.3 在 IDEA 中配置 kafka.Kafka 這個入口類具體配置如下圖所示：5.4 啟動 Kafka Broker啟動成功的話，控制臺輸出沒有異常，且能看到如下輸出：
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 兩個日志包，當然也可以在 build.gradle 中添加對應(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ā)送、消費 message我們這里使用 Kafka 自帶的腳本工具來驗證上面搭建的 Kafka 源碼環(huán)境首先，我們進入到 ${KAFKA_HOME}/bin 目錄，通過 kafka-topics.sh 命令來創(chuàng)建一個名為 topic_test 的 topic：執(zhí)行效果如下圖所示：
然后我們通過 kafka-console-consumer.sh 命令啟動一個命令行的 consumer 來消費 topic_test 這個 topic，如下：
./kafka-console-consumer.sh?--bootstrap-server?localhost:9092?--topic?topic_test
接下來，我們通過 kafka-console-producer.sh 命令啟動一個命令行的 producer 向 topic_test 這個 topic 中生成數(shù)據(jù)，如下：
當我們輸入一條 message 并回車之后，message 會發(fā)送到 topic_test 這個 topic 中。
大功告成，后續(xù)會陸續(xù)分析 Kafka Broker 端的源碼，盡情期待~
下載文檔
