前言

Kafka 是通過 Scala 和 Java共同編寫的語言，之所以選擇2.7.2的版本是因?yàn)檫@個(gè)版本的Kafka是最后一版本保留ZK的版本。

為什么不直接部署最新版代碼？

因?yàn)檫^去很長一段時(shí)間Kafka都是和ZK配合的，并且有很多成熟項(xiàng)目都使用了帶ZK的Kafka，去ZK的Kafka還有不少的路要走。

環(huán)境準(zhǔn)備

• JDK：1.8.0_351

• Scala：2.12.8

• Gradle：6.6

• Zookeeper：3.4.14

Kafka 2.7.2

建議fork一個(gè)官方的分支到自己的倉庫，方便自己學(xué)習(xí)的時(shí)候添加注釋等內(nèi)容。

地址：https://github.com/apache/kafka

安裝JDK 1.8

JDK的教程玩網(wǎng)上有非常多，尋找合適的JDK8的版本即可。

https://www.oracle.com/java/technologies/javase/javase8-archive-downloads.html

安裝 Scala 2.12.8

下載

下載地址為：https://www.scala-lang.org/download/2.12.8.html

個(gè)人為Win 11，直接下載Windows 安裝版本：

個(gè)人是window系統(tǒng)，直接點(diǎn)擊下一步即可，安裝過程這里省略了。

配置 SCALA_HOME

安裝完成之后是在對應(yīng)的操作系統(tǒng)配置環(huán)境變量，個(gè)人在Path變量中增加SCALA_HOME，并且指定地址即可。因?yàn)閭€(gè)人是Windows 安裝版本安裝，已經(jīng)自動(dòng)配置了環(huán)境變量。安裝完成之后結(jié)果如下：

C:\Users\adong>scala?-version
Scala?code?runner?version?2.12.8?--?Copyright?2002-2018,?LAMP/EPFL?and?Lightbend,?Inc.


安裝 gradle 6.6

下載地址：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.1-bin.zip。

配置 Gradle 環(huán)境變量

直接把gradle 的安裝目錄bin地址貼到環(huán)境變量的Path當(dāng)中。

E:\adongstack\mysoft\gradle-6.6.1\bin


然后是在cmd當(dāng)中驗(yàn)證即可：

C:\Users\adong>gradle?-v

Welcome?to?Gradle?6.6.1!

Here?are?the?highlights?of?this?release:
?-?Experimental?build?configuration?caching
?-?Built-in?conventions?for?handling?credentials
?-?Java?compilation?supports?--release?flag

For?more?details?see?https://docs.gradle.org/6.6.1/release-notes.html


------------------------------------------------------------
Gradle?6.6.1
------------------------------------------------------------

Build?time:???2020-08-25?16:29:12?UTC
Revision:?????f2d1fb54a951d8b11d25748e4711bec8d128d7e3

Kotlin:???????1.3.72
Groovy:???????2.5.12
Ant:??????????Apache?Ant(TM)?version?1.10.8?compiled?on?May?10?2020
JVM:??????????1.8.0_333?(Oracle?Corporation?25.333-b02)
OS:???????????Windows?11?10.0?amd64


安裝 Zookeeper 3.4.14

個(gè)人建議安裝到虛擬機(jī)的Linux系統(tǒng)，安裝的操作如下：

wget?https://archive.apache.org/dist/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz


國內(nèi)的訪問速度比較慢，這里直接上外網(wǎng)拖了一個(gè)安裝包放到網(wǎng)盤，讀者根據(jù)網(wǎng)盤下載即可：

鏈接：https://pan.baidu.com/s/1R6zz5pffuwV8D7B-N6PV7Q?pwd=wopn 提取碼：wopn

創(chuàng)建 data 文件夾

在Zookeeper 3.4.14的目錄當(dāng)中準(zhǔn)備data文件夾。

[zxd@localhost?zookeeper-3.4.14]$?mkdir?data
mkdir:?cannot?create?directory?‘data’:?Permission?denied
[zxd@localhost?zookeeper-3.4.14]$?sudo?mkdir?data
[sudo]?password?for?zxd:?
[zxd@localhost?zookeeper-3.4.14]$?ls
bin??conf??data??dist-maven


修改 zoo.cfg 中的 data 屬性

我們將Zookeeper的數(shù)據(jù)目錄執(zhí)行剛剛新建的data文件夾。

cd?conf
vim?zoo_sample.cfg?zoo.cfg


vim zoo_sample.cfg zoo.cfg，我們修改相關(guān)的配置，默認(rèn)目錄在/tmp/zookeeper，這里個(gè)人改為/opt/zookeeper/zookeeper-3.4.14/data

#?the?directory?where?the?snapshot?is?stored.
#?do?not?use?/tmp?for?storage,?/tmp?here?is?just?
#?example?sakes.
#dataDir=/tmp/zookeeper
dataDir=/opt/zookeeper/zookeeper-3.4.14/data
#?the?port?at?which?the?clients?will?connect
clientPort=2181


修改之后我們使用-x 應(yīng)用修改。

啟動(dòng)Zookeeper

編輯完成之后，進(jìn)入到安裝目錄的bin目錄當(dāng)中啟動(dòng)Zookeeper。

[zxd@localhost?zookeeper-3.4.14]$?cd?bin/
[zxd@localhost?bin]$?sudo?./zkServer.sh?start?
ZooKeeper?JMX?enabled?by?default
Using?config:?/opt/zookeeper/zookeeper-3.4.14/bin/../conf/zoo.cfg
grep:?/opt/zookeeper/zookeeper-3.4.14/bin/../conf/zoo.cfg:?No?such?file?or?directory
mkdir:?cannot?create?directory?‘’:?No?such?file?or?directory
Starting?zookeeper?...?STARTED


發(fā)現(xiàn)沒有啟動(dòng)成功，根據(jù)提示我們需要zoo.cfg的配置，這部分為Zk的內(nèi)容，這里就直接跳過了。

Kafka 2.7.2 源碼下載

下載kafka-2.7.2-src.tgz包之后，解壓到對應(yīng)的目錄，接下來需要導(dǎo)入到Idea當(dāng)中。

導(dǎo)入Idea

下載源碼之后在Idea當(dāng)中導(dǎo)入Kafka項(xiàng)目。

導(dǎo)入項(xiàng)目之后，我們進(jìn)入到Setting， 設(shè)置Gradle的Home地址。

這里把Gradle的Home地址設(shè)置為自己本機(jī)安裝的，不使用Iead自帶的。

修改 build.gradle

接下來還不能導(dǎo) jar 包，需要把鏡像文件下載服務(wù)器更換為國內(nèi)的私服，否則會(huì)相當(dāng)慢，直接導(dǎo)致 "time out" 報(bào)錯(cuò)。進(jìn)入 kafka 源碼包，修改 build.gradle 文件，在原來配置上，添加阿里的私服配置。

下面的內(nèi)容復(fù)制到 build.gradle 文件的對應(yīng)位置：

maven?{
?url?'http://maven.aliyun.com/nexus/content/groups/public/'
}
maven?{
?url?'http://maven.aliyun.com/nexus/content/repositories/jcenter'
}


文件當(dāng)中查找buildscript可以快速定位。

注意最后需要 調(diào)整一下位置，優(yōu)先從阿里云的庫獲取：

buildscript?{??
??repositories?{??
????maven?{??
??????url?'http://maven.aliyun.com/nexus/content/groups/public/'??
????}??
????maven?{??
??????url?'http://maven.aliyun.com/nexus/content/repositories/jcenter'??
????}??
????mavenCentral()??
????jcenter()??
????maven?{??
??????url?"https://plugins.gradle.org/m2/"??
????}??
??
??}
??......


接著還需要復(fù)制到allproject當(dāng)中：

maven?{
?url?'http://maven.aliyun.com/nexus/content/groups/public/'
}
maven?{
?url?'http://maven.aliyun.com/nexus/content/repositories/jcenter'
}


復(fù)制到的位置大概如下：

allprojects?{??
??
??repositories?{??
????maven?{??
??????url?'http://maven.aliyun.com/nexus/content/groups/public/'??
????}??
????maven?{??
??????url?'http://maven.aliyun.com/nexus/content/repositories/jcenter'??
????}??
????mavenCentral()??
??
??}


替換之后耐心等待依賴下載完成：

在依賴安裝完成之后，我們可以點(diǎn)一下Idea頂部的小錘子，最后的提示編譯成功：

安裝 Idea scala 插件

Idea導(dǎo)入源碼之后的.scala文件是沒有語法提示的，我們安裝插件之后可以正常閱讀scala的源碼：

插件安裝完成之后需要重啟。訪問kafka.scala，如果相關(guān)的代碼出現(xiàn)顏色和語法提示說明插件安裝成功：

選擇 scala jdk

如果讀者是第一次搭建Kafka和使用scala，大概率會(huì)出現(xiàn)下面的提示：

根據(jù)提示設(shè)置scala jdk的安裝目錄：

C:\Program?Files?(x86)\scala\bin


設(shè)置完成之后一切就大功告成了。

編譯和構(gòu)建 Kafka 源碼

這里個(gè)人不太熟悉Idea對于gradle的使用，最后用了gradle的命令進(jìn)行構(gòu)建，構(gòu)建之后出現(xiàn)下面的內(nèi)容表示正確：

gradle命令是用來下載和更新 Gradle 套件（Gradle Wrapper）的，gradle jar是用 Gradle 套件構(gòu)建 Kafka 工程，生成 Jar 文件。

下面是使用gradle 命令的結(jié)果：

adong@Adong-pc?MINGW64?/e/adongstack/project/selfUp/kafka-2.7/kafka-2.7.2-src
$?gradle
Starting?a?Gradle?Daemon,?1?busy?Daemon?could?not?be?reused,?use?--status?for?details

>?Configure?project?:
Building?project?'core'?with?Scala?version?2.13.3
Building?project?'streams-scala'?with?Scala?version?2.13.3

>?Task?:help

Welcome?to?Gradle?6.6.1.

To?run?a?build,?run?gradle?<task>?...

To?see?a?list?of?available?tasks,?run?gradle?tasks

To?see?a?list?of?command-line?options,?run?gradle?--help

To?see?more?detail?about?a?task,?run?gradle?help?--task?<task>

For?troubleshooting,?visit?https://help.gradle.org

Deprecated?Gradle?features?were?used?in?this?build,?making?it?incompatible?with?Gradle?7.0.
Use?'--warning-mode?all'?to?show?the?individual?deprecation?warnings.
See?https://docs.gradle.org/6.6.1/userguide/command_line_interface.html#sec:command_line_warnings

BUILD?SUCCESSFUL?in?44s
1?actionable?task:?1?executed



下面使用gradle jar的編譯結(jié)果：

adong@Adong-pc?MINGW64?/e/adongstack/project/selfUp/kafka-2.7/kafka-2.7.2-src
$?gradle?jar

>?Configure?project?:
Building?project?'core'?with?Scala?version?2.13.3
Building?project?'streams-scala'?with?Scala?version?2.13.3

>?Task?:clients:processMessages
MessageGenerator:?processed?121?Kafka?message?JSON?files(s).

>?Task?:raft:processMessages
MessageGenerator:?processed?1?Kafka?message?JSON?files(s).
<=------------>?12%?EXECUTING?[2m?6s]

>?Task?:clients:processTestMessages
MessageGenerator:?processed?2?Kafka?message?JSON?files(s).
<==----------->?21%?EXECUTING?[3m?10s]

>?Task?:streams:processMessages
MessageGenerator:?processed?1?Kafka?message?JSON?files(s).

Deprecated?Gradle?features?were?used?in?this?build,?making?it?incompatible?with?Gradle?7.0.
Use?'--warning-mode?all'?to?show?the?individual?deprecation?warnings.
See?https://docs.gradle.org/6.6.1/userguide/command_line_interface.html#sec:command_line_warnings

BUILD?SUCCESSFUL?in?3m?58s
87?actionable?tasks:?87?executed



最后使用下面的命令：

gradle?clean?build?-x?test


但是這種命令構(gòu)建過程會(huì)出現(xiàn)一些報(bào)錯(cuò)，可以把報(bào)錯(cuò)所需要的依賴jar包放到下面：

E:\adongstack\project\selfUp\kafka-2.7\kafka-2.7.2-src\gradle\wrapper


要構(gòu)建整個(gè) Kafka 工程并打包出一個(gè)可運(yùn)行的二進(jìn)制環(huán)境命令如下

gradle?clean?releaseTarGz


其他構(gòu)建命令

還有其他gradle的構(gòu)建命令：

構(gòu)建?jar包并運(yùn)行
./gradle?jar

構(gòu)建項(xiàng)目，看你是idea工具還是eclipse
./gradle?idea
./gradle?eclipse

構(gòu)建源碼包
./gradle?srcJar

構(gòu)建javadoc文檔
./gradle?aggregatedJavadoc

清理并構(gòu)建
./gradle?clean


單元測試

Kafka當(dāng)中同樣存在很多單元測試，下面是一些核心模塊的單元測試命令。

core 模塊單元測試：

gradle?core:test


client 模塊單元測試：

gradle?clients:test


模塊當(dāng)中的某個(gè)子模塊單元測試：

gradle?connect:[submodule]:test


Connect 工程下細(xì)分了多個(gè)子模塊，比如 api、runtime 等，需要顯式地指定要 測試的子模塊名才能進(jìn)行測試。

stream 模塊單元測試。

gradle?connect:[submodule]:test


個(gè)人實(shí)際使用之后會(huì)發(fā)現(xiàn)有部分報(bào)錯(cuò)信息：

除了上面的整個(gè)模塊的單元測試，如果只想要測試某個(gè)模塊下的某個(gè)類，可以使用下面的方法：

單獨(dú)對某一個(gè)具體的測試用例進(jìn)行測試，比如測試 LogTest 類：

gradle?core:test?--tests?kafka.log.LogTest


個(gè)人的執(zhí)行結(jié)果如下：

$?gradle?core:test?--tests?kafka.log.LogTest

>?Configure?project?:
Building?project?'core'?with?Scala?version?2.13.3
Building?project?'streams-scala'?with?Scala?version?2.13.3

>?Task?:core:test
kafka.log.LogTest.testLogDelete?failed,?log?available?in?E:\adongstack\project\selfUp\kafka-2.7\kafka-2.7.2-src\core\build\reports\testOutput\kafka.log.LogTest.testLogDelete.test.stdout

kafka.log.LogTest?>?testLogDelete?FAILED
????org.apache.kafka.common.errors.KafkaStorageException:?Error?while?deleting?log?for?kafka-936488?in?dir?C:\Users\adong\AppData\Local\Temp\kafka-7162272829353105740

????????Caused?by:
????????java.nio.file.FileSystemException:?C:\Users\adong\AppData\Local\Temp\kafka-7162272829353105740\kafka-936488\00000000000000000000.timeindex?->?C:\Users\adong\AppData\Local\Temp\kafka-7162272829353105740\kafka-936488\00000000000000000000.timeindex.deleted:???????????????????????????????????



Kafka目錄結(jié)構(gòu)

bin 目錄：保存 Kafka 工具行腳本，我們熟知的 kafka-server-start 和 kafka-consoleproducer 等腳本都存放在這。

clients 目錄：保存 Kafka 客戶端代碼，比如生產(chǎn)者和消費(fèi)者的代碼都在該目錄下。

config 目錄：保存 Kafka 的配置文件，其中比較重要的配置文件是 server.properties。

connect 目錄：保存 Connect 組件的源代碼。Kafka Connect 組件是用來實(shí)現(xiàn) Kafka 與外部系統(tǒng)之間的實(shí)時(shí)數(shù)據(jù)傳輸的。

core 目錄：保存 Broker 端代碼。Kafka 服務(wù)器端代碼全部保存在該目錄下。

streams 目錄：保存 Streams 組件的源代碼。Kafka Streams 是實(shí)現(xiàn) Kafka 實(shí)時(shí)流處理的組件。

server 目錄：顧名思義，它是 Kafka 的服務(wù)器端主代碼，里面的類非常多，很多關(guān)鍵的 Kafka 組件都存放在這里，比如狀態(tài)機(jī)、Purgatory 延時(shí)機(jī)制等。

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ò)的方式來做這件事情。

coordinator 包：保存了消費(fèi)者端的 GroupCoordinator 代碼和用于事務(wù)的 TransactionCoordinator 代碼。對 coordinator 包進(jìn)行分析，特別是對消費(fèi)者端的 GroupCoordinator 代碼進(jìn)行分析，是我們弄明白 Broker 端協(xié)調(diào)者組件設(shè)計(jì)原理的關(guān) 鍵。

其他目錄結(jié)構(gòu)

除了上面的核心目錄之外，我們還可以在項(xiàng)目根路徑看到一些其他模塊：

network 包：封裝了 Kafka 服務(wù)器端網(wǎng)絡(luò)層的代碼，特別是 SocketServer.scala 這個(gè)文件，是 Kafka 實(shí)現(xiàn) Reactor 模式的具體操作類，非常值得一讀。

consumer 包：后面會(huì)丟棄該包，用 clients 包下 consumer 相關(guān)類代替。

tools 包：工具類。

log 包：保存了 Kafka 最核心的日志結(jié)構(gòu)代碼，包括日志、日志段、索引文件等, 另外，該包下還封裝了 Log Compaction 的實(shí)現(xiàn)機(jī)制，是非常重要的源碼包。

checkstyle 目錄：代碼規(guī)范，自動(dòng)化檢測。

Checkstyle 是什么，類似于代碼規(guī)范的自動(dòng)化檢測插件，國內(nèi)最為經(jīng)典的是阿里巴巴的規(guī)約插件，而最出名的檢查插件為 google style guide，Checkstyle 就是以類似這種風(fēng)格開發(fā)出的一個(gè)自動(dòng)化插件，來輔助判斷代碼格式是否滿足規(guī)范。

該目錄下的文件定義了工程代碼格式的規(guī)范，我們可以在 build.gradle 中看到相關(guān) checkstyle 的配置和自動(dòng)化代碼格式化配置：

checkstyle 配置（build.gradle）

def?checkstyleConfigProperties(configFileName)?{??
??[importControlFile:?"$rootDir/checkstyle/$configFileName",??
???suppressionsFile:?"$rootDir/checkstyle/suppressions.xml",??
???headerFile:?"$rootDir/checkstyle/java.header"]??
}

checkstyle?{??
??configProperties?=?checkstyleConfigProperties("import-control-core.xml")??
}


connect 目錄：保存 Connect 組件的源代碼。 Kafka Connect 組件是用來實(shí)現(xiàn) Kafka 與外部系統(tǒng)之間的實(shí)時(shí)數(shù)據(jù)傳輸?shù)摹?/p>

docs 目錄：Kafka 設(shè)計(jì)文檔以及組件相關(guān)結(jié)構(gòu)圖。

examples 目錄：Kafka 樣例相關(guān)目錄。

gradle 目錄：gradle 的腳本和依賴包定義等相關(guān)文件。

jmh-benchmarks 目錄：Kafka 代碼微基準(zhǔn)測試相關(guān)類。

JMH，即 Java Microbenchmark Harness，是專門用于代碼微基準(zhǔn)測試的工具套件。Micro Benchmark 是什么意思？簡單的來說就是基于方法層面的基準(zhǔn)測試，精度可以達(dá)到微秒級。當(dāng)你定位到熱點(diǎn)方法，希望進(jìn)一步優(yōu)化方法性能的時(shí)候，就可以使用 JMH 對優(yōu)化的結(jié)果進(jìn)行量化的分析。

JMH 比較典型的應(yīng)用場景有：

• 準(zhǔn)確的知道某個(gè)方法需要執(zhí)行多長時(shí)間，以及執(zhí)行時(shí)間和輸入之間的相關(guān)性；

• 對比接口不同實(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)模塊。

tests 目錄：此目錄的內(nèi)容介紹如何進(jìn)行 Kafka 系統(tǒng)集成和性能測試。

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)境。

參考

https://segmentfault.com/a/1190000040790524