當(dāng)前位置:首頁 > 公眾號(hào)精選 > 架構(gòu)師社區(qū)
[導(dǎo)讀]一、前言老周這里編譯Kafka的版本是2.7,為啥采用這個(gè)版本來搭建源碼的閱讀環(huán)境呢?因?yàn)樵摪姹鞠鄬?duì)來說比較新。而我為啥不用2.7后的版本呢?比如2.8,這是因?yàn)槿サ袅薢ooKeeper,還不太穩(wěn)定,生產(chǎn)環(huán)境也不太建議使用,所以以2.7版本進(jìn)行源碼搭建并研究。二、環(huán)境準(zhǔn)備JDK:...

一、前言

老周這里編譯 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

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境
這里老周是 Mac OS 系統(tǒng),這里大家看著自己的系統(tǒng)來下就好了哈。

3.2.1 配置 Scala 環(huán)境變量

終端輸入以下命令進(jì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 驗(yàn)證

終端輸入以下命令:

scala?-version
出現(xiàn)以下提示,說明 Scala 環(huán)境搭建成功。

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境
3.3 Gradle 環(huán)境搭建

首先來到 Gradle官網(wǎng):https://services.gradle.org/distributions/

如下圖:

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

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

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境
3.3.1 ?配置 Gradle 環(huán)境變量

終端輸入以下命令進(jì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 驗(yàn)證

終端輸入以下命令:

gradle?-v
出現(xiàn)以下提示,說明 Gradle 環(huán)境搭建成功。

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(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)成功

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境
3.5 Kafka 源碼環(huán)境搭建

官網(wǎng)下載對(duì)應(yīng)版本的源碼包,網(wǎng)址:http://kafka.apache.org/downloads

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

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

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

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'
????????}
????}
}
聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

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 提取碼: hpj5

gradle 其它命令:

#?構(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)

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境
4.1 ?代碼安裝包結(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 配置:

    聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境


    聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境


    scala 自動(dòng)化代碼格式化配置:


    聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

  • 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:

    聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

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

4.2 項(xiàng)目結(jié)構(gòu)

項(xiàng)目結(jié)構(gòu)的話主要關(guān)注 core 目錄,core 目錄 是 Kafka 核心包,有集群管理,分區(qū)管理,存儲(chǔ)管理,副本管理,消費(fèi)者組管理,網(wǎng)絡(luò)通信,消費(fèi)管理等核心類。

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(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 目錄下。

如下圖所示:

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境


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è)入口類

具體配置如下圖所示:

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

5.4 啟動(dòng) Kafka Broker啟動(dòng)成功的話,控制臺(tái)輸出沒有異常,且能看到如下輸出:

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境
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 兩個(gè)日志包,當(dāng)然也可以在 build.gradle 中添加對(duì)應(yīng)的配置來添加包。

方法1:

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

方法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 模塊:

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

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.
聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

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:編譯?Kafka?源碼并搭建源碼環(huán)境

然后我們通過 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:編譯?Kafka?源碼并搭建源碼環(huán)境


接下來,我們通過 kafka-console-producer.sh 命令啟動(dòng)一個(gè)命令行的 producer 向 topic_test 這個(gè) topic 中生成數(shù)據(jù),如下:
聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

當(dāng)我們輸入一條 message 并回車之后,message 會(huì)發(fā)送到 topic_test 這個(gè) topic 中。
聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境
我們輸入完 message 并回車之后,就可以在 consumer 處收到該 message 了,效果如下圖所示:

聊聊?Kafka:編譯?Kafka?源碼并搭建源碼環(huán)境

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

本站聲明: 本文章由作者或相關(guān)機(jī)構(gòu)授權(quán)發(fā)布,目的在于傳遞更多信息,并不代表本站贊同其觀點(diǎn),本站亦不保證或承諾內(nèi)容真實(shí)性等。需要轉(zhuǎn)載請(qǐng)聯(lián)系該專欄作者,如若文章內(nèi)容侵犯您的權(quán)益,請(qǐng)及時(shí)聯(lián)系本站刪除。
換一批
延伸閱讀

9月2日消息,不造車的華為或?qū)⒋呱龈蟮莫?dú)角獸公司,隨著阿維塔和賽力斯的入局,華為引望愈發(fā)顯得引人矚目。

關(guān)鍵字: 阿維塔 塞力斯 華為

倫敦2024年8月29日 /美通社/ -- 英國(guó)汽車技術(shù)公司SODA.Auto推出其旗艦產(chǎn)品SODA V,這是全球首款涵蓋汽車工程師從創(chuàng)意到認(rèn)證的所有需求的工具,可用于創(chuàng)建軟件定義汽車。 SODA V工具的開發(fā)耗時(shí)1.5...

關(guān)鍵字: 汽車 人工智能 智能驅(qū)動(dòng) BSP

北京2024年8月28日 /美通社/ -- 越來越多用戶希望企業(yè)業(yè)務(wù)能7×24不間斷運(yùn)行,同時(shí)企業(yè)卻面臨越來越多業(yè)務(wù)中斷的風(fēng)險(xiǎn),如企業(yè)系統(tǒng)復(fù)雜性的增加,頻繁的功能更新和發(fā)布等。如何確保業(yè)務(wù)連續(xù)性,提升韌性,成...

關(guān)鍵字: 亞馬遜 解密 控制平面 BSP

8月30日消息,據(jù)媒體報(bào)道,騰訊和網(wǎng)易近期正在縮減他們對(duì)日本游戲市場(chǎng)的投資。

關(guān)鍵字: 騰訊 編碼器 CPU

8月28日消息,今天上午,2024中國(guó)國(guó)際大數(shù)據(jù)產(chǎn)業(yè)博覽會(huì)開幕式在貴陽舉行,華為董事、質(zhì)量流程IT總裁陶景文發(fā)表了演講。

關(guān)鍵字: 華為 12nm EDA 半導(dǎo)體

8月28日消息,在2024中國(guó)國(guó)際大數(shù)據(jù)產(chǎn)業(yè)博覽會(huì)上,華為常務(wù)董事、華為云CEO張平安發(fā)表演講稱,數(shù)字世界的話語權(quán)最終是由生態(tài)的繁榮決定的。

關(guān)鍵字: 華為 12nm 手機(jī) 衛(wèi)星通信

要點(diǎn): 有效應(yīng)對(duì)環(huán)境變化,經(jīng)營(yíng)業(yè)績(jī)穩(wěn)中有升 落實(shí)提質(zhì)增效舉措,毛利潤(rùn)率延續(xù)升勢(shì) 戰(zhàn)略布局成效顯著,戰(zhàn)新業(yè)務(wù)引領(lǐng)增長(zhǎng) 以科技創(chuàng)新為引領(lǐng),提升企業(yè)核心競(jìng)爭(zhēng)力 堅(jiān)持高質(zhì)量發(fā)展策略,塑強(qiáng)核心競(jìng)爭(zhēng)優(yōu)勢(shì)...

關(guān)鍵字: 通信 BSP 電信運(yùn)營(yíng)商 數(shù)字經(jīng)濟(jì)

北京2024年8月27日 /美通社/ -- 8月21日,由中央廣播電視總臺(tái)與中國(guó)電影電視技術(shù)學(xué)會(huì)聯(lián)合牽頭組建的NVI技術(shù)創(chuàng)新聯(lián)盟在BIRTV2024超高清全產(chǎn)業(yè)鏈發(fā)展研討會(huì)上宣布正式成立。 活動(dòng)現(xiàn)場(chǎng) NVI技術(shù)創(chuàng)新聯(lián)...

關(guān)鍵字: VI 傳輸協(xié)議 音頻 BSP

北京2024年8月27日 /美通社/ -- 在8月23日舉辦的2024年長(zhǎng)三角生態(tài)綠色一體化發(fā)展示范區(qū)聯(lián)合招商會(huì)上,軟通動(dòng)力信息技術(shù)(集團(tuán))股份有限公司(以下簡(jiǎn)稱"軟通動(dòng)力")與長(zhǎng)三角投資(上海)有限...

關(guān)鍵字: BSP 信息技術(shù)
關(guān)閉
關(guān)閉