聊聊 Kafka:編譯 Kafka 源碼并搭建源碼環(huán)境
一、前言
老周這里編譯 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
這里老周是 Mac OS 系統(tǒng)，這里大家看著自己的系統(tǒng)來下就好了哈。
3.2.1 配置 Scala 環(huán)境變量
終端輸入以下命令進行編輯：
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)境搭建成功。
3.3 Gradle 環(huán)境搭建
首先來到 Gradle官網(wǎng)：https://services.gradle.org/distributions/
如下圖：
我們選擇想要安裝的發(fā)布版本，gradle-x.x-bin.zip 是需要下載的安裝發(fā)布版，gradle-x.x-src.zip 是源碼，gradle-x.x-all.zip 則是下載全部的文件。我本地為 gradle-6.6。
Gradle下載的源碼不需要安裝，我們將下載的壓縮包在本機的目錄下直接解壓即可，解壓后的目錄如下圖所示。
3.3.1 ?配置 Gradle 環(huán)境變量
終端輸入以下命令進行編輯：
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 服務啟動
進入 bin 目錄，啟動服務輸入命令
./zkServer.sh?start
輸出以下內(nèi)容表示啟動成功
3.5 Kafka 源碼環(huán)境搭建
官網(wǎng)下載對應版本的源碼包，網(wǎng)址：http://kafka.apache.org/downloads
下載完后解壓，這個源碼文件還需要導入依賴 jar 包，個人使用 IDEA 來 import 導入項目，導入完后需使用前面配置好的 gradle 作為 Gradle home 地址。
3.5.1 導入 Kafka 源碼至 IDEA
3.5.2 修改 build.gradle
接下來還不能導 jar 包，需要把鏡像文件下載服務器更換為國內(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 提取碼: hpj5
gradle 其它命令：
#?構(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)
4.1 ?代碼安裝包結(jié)構(gòu)
bin 目錄：保存 Kafka 工具行腳本，我們熟知的 kafka-server-start 和 kafka-console-producer 等腳本都存放在這里。
checkstyle 目錄：代碼規(guī)范，自動化檢測。
Checkstyle 是什么，關于格式化的討論就不曾中斷過，到底什么才是正確的，什么才是錯誤的，到現(xiàn)在也沒有完整的定論。但隨著時間發(fā)展，漸漸衍生出一套規(guī)范出來。沒有什么絕對的正確和錯誤，關鍵在于規(guī)范的定義。最出名的就是 google style guide，Checkstyle 就是以這種風格開發(fā)出的一個自動化插件，來輔助判斷代碼格式是否滿足規(guī)范。
該目錄下的文件定義了工程代碼格式的規(guī)范，我們可以在 build.gradle 中看到相關 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 服務器端代碼全部保存在該目錄下。
docs 目錄：Kafka 設計文檔以及組件相關結(jié)構(gòu)圖。
examples 目錄：Kafka 樣例相關目錄。
generator 目錄：Kafka 消息類處理模塊，主要是根據(jù) clients 模塊下的 message json 文件生成對應的 java 類，在 build.gradle 文件中，可以看到定義了一個任務 processMessages：
gradle 目錄：gradle 的腳本和依賴包定義等相關文件。
jmh-benchmarks 目錄：Kafka 代碼微基準測試相關類。
JMH，即 Java Microbenchmark Harness，是專門用于代碼微基準測試的工具套件。何謂 Micro Benchmark 呢？簡單的來說就是基于方法層面的基準測試，精度可以達到微秒級。當你定位到熱點方法，希望進一步優(yōu)化方法性能的時候，就可以使用 JMH 對優(yōu)化的結(jié)果進行量化的分析。
JMH 比較典型的應用場景有：
想準確的知道某個方法需要執(zhí)行多長時間，以及執(zhí)行時間和輸入之間的相關性；
對比接口不同實現(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é)議相關。
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)用，整個應用的運行方式主要由開發(fā)者控制，方便使用和調(diào)試。
Kafka Streams 是一個用來構(gòu)建流處理程序的庫，特別是其輸入是一個 Kafka topic，輸出是另一個 Kafka topic 的程序（或者是調(diào)用外部服務，或者是更新數(shù)據(jù)庫，或者其它）。它使得你以一種分布式以及容錯的方式來做這件事情。
tests 目錄：此目錄的內(nèi)容介紹如何進行 Kafka 系統(tǒng)集成和性能測試。
tools 目錄：工具類模塊。
vagrant 目錄：介紹如何在 Vagrant 虛擬環(huán)境中運行 Kafka，提供了相關的腳本文件和說明文檔。
Vagrant 是一個基于 Ruby 的工具，用于創(chuàng)建和部署虛擬化開發(fā)環(huán)境。它使用 Oracle 的開源 VirtualBox 虛擬化系統(tǒng)，使用 Chef 創(chuàng)建自動化虛擬環(huán)境。
4.2 項目結(jié)構(gòu)
項目結(jié)構(gòu)的話主要關注 core 目錄，core 目錄 是 Kafka 核心包，有集群管理，分區(qū)管理，存儲管理，副本管理，消費者組管理，網(wǎng)絡通信，消費管理等核心類。
admin 包：執(zhí)行管理命令的功能；
api 包：封裝請求和響應 DTO 對象；
cluster 包：集群對象，例如 Replica 類代表一個分區(qū)副本，Partition 類代表一個分區(qū)；
common 包：通用 jar 包；
controller包：和kafkaController（kc）相關的類，重點模塊，一個kafka集群只有一個leader kc，該kc負責 分區(qū)管理，副本管理，并保證集群信息在集群中同步；
coordinator 包：保存了消費者端的 GroupCoordinator 代碼和用于事務的 TransactionCoordinator 代碼。對 coordinator 包進行分析，特別是對消費者端的 GroupCoordinator 代碼進行分析，是 Broker 端協(xié)調(diào)者組件設計原理的關鍵。
log 包：保存了 Kafka 最核心的日志結(jié)構(gòu)代碼，包括日志、日志段、索引文件等, 另外，該包下還封裝了 Log Compaction 的實現(xiàn)機制，是非常重要的源碼包。
network 包：封裝了 Kafka 服務器端網(wǎng)絡層的代碼，特別是 SocketServer.scala 這個文件，是 Kafka 實現(xiàn) Reactor 模式的具體操作類，非常值得一讀。
consumer 包：后面會丟棄該包，用 clients 包下 consumer 相關類代替。
server 包：顧名思義，它是 Kafka 的服務器端主代碼，里面的類非常多，很多關鍵的 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
啟動成功的話，控制臺輸出沒有異常，且能看到如下輸出：
5.5 可能出現(xiàn)以下異常
5.5.1 異常1
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 中添加對應的配置來添加包。
方法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 中。
我們輸入完 message 并回車之后，就可以在 consumer 處收到該 message 了，效果如下圖所示：
大功告成，后續(xù)會陸續(xù)分析 Kafka Broker 端的源碼，盡情期待~
歡迎大家關注我的公眾號【老周聊架構(gòu)】，Java后端主流技術棧的原理、源碼分析、架構(gòu)以及各種互聯(lián)網(wǎng)高并發(fā)、高性能、高可用的解決方案。