我在低級別上使用過很多 Apache Kafka 協(xié)議。僅按照官方指南開始執(zhí)行此操作并不容易，而且我閱讀了很多代碼。通過這篇文章，我想一步步指導(dǎo)您從原始值到有意義的請求，為您提供一個良好的開端。

在這篇文章中：

1. 探索 Kafka 協(xié)議代碼以及使用 Wireshark 運行的協(xié)議。
2. 學(xué)習(xí)如何讀取和寫入原始值。
3. 組合基元來執(zhí)行有意義的請求。

我們將使用Python作為編程語言。但是，代碼將是零依賴性的，并且可以輕松移植到您選擇的語言。

簡介

Apache Kafka 有一個自定義的二進制協(xié)議，該協(xié)議是有版本的，具有各種數(shù)據(jù)類型、可選字段等。不幸的是，它沒有使用像 Protobuf 這樣眾所周知的序列化格式。協(xié)議消息架構(gòu)以 JSON 格式描述。執(zhí)行序列化和反序列化的實際 Java 代碼是根據(jù)此描述生成的。

當(dāng)你身處Java世界時，你可以使用官方的客戶端庫。但如果您使用其他平臺，則需要依賴第三方實現(xiàn)。它們存在，但主要關(guān)注生產(chǎn)者和消費者，很少關(guān)注管理客戶端的某些方面。如果您需要做其他事情，您就得靠自己了。

這篇文章將幫助您開始破解 Kafka 協(xié)議。 （如果您正在為 Kafka 協(xié)議尋找現(xiàn)成的 Python（反）序列化庫，請查看 Kio¹。對于 Rust，請查看我正在開發(fā)的庫。）

您可以在 Github 上的這個存儲庫中找到這篇文章中的代碼以及更多類似的測試。

協(xié)議概述

您可以在此頁面找到官方協(xié)議說明。我鼓勵您熟悉它，至少閱讀“預(yù)備知識”和“協(xié)議”部分。

以下是一些亮點。 Kafka 協(xié)議是基于 TCP 的二進制請求-響應(yīng)協(xié)議：

• 基于 TCP：Kafka 代理偵聽 TCP 堆棧上的端口（這提供了一些好處，例如排序保證）。
• 二進制：消息以二進制形式編碼，需要根據(jù)預(yù)定義的模式進行特殊的序列化和反序列化。
• 請求-響應(yīng)：交換由客戶端發(fā)起，服務(wù)器端是被動的，只回復(fù)請求。

每種 API 消息類型都由請求和響應(yīng)對組成，并由稱為 API 密鑰的數(shù)值進行標(biāo)識。例如，Kafka 最具特色的 RPC 的 Produce 和 Fetch 對應(yīng)的 API 密鑰為 0 和 1。如今，API 消息類型接近 90 種（其中一些是經(jīng)紀(jì)商間的消息類型，而不是客戶經(jīng)紀(jì)商間的消息類型）。

請求和響應(yīng)由版本化模式描述。版本控制允許協(xié)議演變，例如添加或刪除字段或更改其數(shù)據(jù)類型。

第一步

您可以執(zhí)行以下一些操作來開始使用 Kafka 協(xié)議。

學(xué)習(xí)Kafka協(xié)議代碼

Kafka 代碼是該協(xié)議（實際上）的真相來源。從 Github 查看 Kafka 代碼并切換到您感興趣的版本（例如 3.8.0）：

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

您可以在clients/src/main/resources/common/message中找到JSON格式的API消息定義。每個 JSON 文件包含一條消息² 類型及其所有版本的定義。 client/src/main/resources/common/message/README.md 很好地概述了模式定義格式。注意默認(rèn)值、靈活版本和標(biāo)記字段等內(nèi)容。

除了您感興趣的具體API消息類型之外，還可以查看clients/src/main/resources/common/message/RequestHeader.json和ResponseHeader.json，它們描述了每個請求-響應(yīng)交換中使用的標(biāo)頭.

讓我們運行代碼生成器：

./gradlew processMessages

現(xiàn)在你可以在clients/src/ generated/java/org/apache/kafka/common/message中找到生成的類。

看看clients/src/ generated/java/org/apache/kafka/common/message/ApiMessageType.java。該實用程序：

• 描述了整套現(xiàn)有的 API 消息類型及其架構(gòu)和版本；
• 將API消息版本映射到requestHeaderVersion和responseHeaderVersion函數(shù)中的請求和響應(yīng)標(biāo)頭版本。

其他文件是從相應(yīng)的架構(gòu) JSON 一對一生成的（有時帶有數(shù)據(jù)后綴，這是一個兼容性問題）。在這些文件中，您會發(fā)現(xiàn)：

1. 版本化模式定義 SCHEMA_0、SCHEMA_1 等。有時模式在版本之間保持相同。這是正常的，意味著只有請求-響應(yīng)對應(yīng)部分發(fā)生了變化。
2. 讀取和寫入方法，您可以在其中找到協(xié)議序列化和反序列化的基本事實。

注意內(nèi)部類，它們代表了消息的復(fù)雜結(jié)構(gòu)。

在 Docker 中運行 Kafka

在 Docker 中運行 Kafka 是一種讓代理運行來測試協(xié)議或捕獲網(wǎng)絡(luò)交換的便捷方法。從 3.7.0 版本開始，Kafka 團隊構(gòu)建了官方 Docker 鏡像，您可以通過以下方式運行：

docker run --rm -ti -p 9092:9092 apache/kafka:3.8.0

如果您對舊版本感興趣，請在 Docker Hub 中搜索其他鏡像。然而，考慮到 Kafka 協(xié)議是向后和向前兼容的，這可能是不需要的：新的代理將很好地識別舊的協(xié)議版本，并且舊的客戶端可以與新的代理進行通信。

如果您閱讀本文，您的計算機上可能已經(jīng)安裝了 Kafka 命令行工具，但為了以防萬一，您也可以在 Docker 中運行它們。例如，運行此命令來創(chuàng)建一個主題：

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

使用 Wireshark 檢查協(xié)議

熟悉了 Kafka 代碼后，讓我們看看實際的協(xié)議。 Wireshark 是此類檢查廣泛使用的工具。它可以剖析 Kafka 協(xié)議（如果您的版本足夠新，則支持最新版本）。

我從4.5.0版本的源代碼構(gòu)建了Wireshark，因為我的操作系統(tǒng)包很舊，無法用新版本解析Kafka協(xié)議。 Wireshark 4.5.0 應(yīng)該主要支持 Kafka 3.7 協(xié)議版本。不過，您可以嘗試可用的版本，看看它如何適合您。

讓我們在環(huán)回接口上運行 Wireshark，使用端口 9092 捕獲過濾器 (1) 和 kafka 顯示過濾器 (2)：

創(chuàng)建一個主題，看看 Wireshark 向我們展示了什么：

./gradlew processMessages

顯示過濾器刪除所有不相關(guān)的內(nèi)容，只留下 Kafka 請求和響應(yīng)。由于 Wireshark 可以理解協(xié)議中的大多數(shù)消息版本（當(dāng)然取決于 Wireshark 版本），因此您可以方便地查看每個消息的結(jié)構(gòu)。 Wireshark 也會顯示相應(yīng)的字節(jié)。

Wireshark 是一個很棒的調(diào)試工具，可以幫助您了解協(xié)議在特定情況下如何工作以及您的實現(xiàn)存在什么問題。

讀取和寫入原始值

該協(xié)議定義了許多原始類型，您可以在此處找到完整的描述。讓我們?yōu)樗鼈儗崿F(xiàn)讀寫代碼。你可以在這個文件中找到所有的函數(shù)，也可以查看相應(yīng)的測試文件。

固定長度整數(shù)值：INT8、INT16、INT32、INT64 和 UINT16

這些是已知固定長度的整數(shù)：1、2、4 或 8 字節(jié)。當(dāng)然，您可以在整個協(xié)議中找到很多這樣的字段。在本課程中，您可能會（簡單地）看到他們的讀寫是如何在 Kafka 中實現(xiàn)的。

我們首先定義從緩沖區(qū)讀取確切字節(jié)數(shù)的函數(shù)³:

docker run --rm -ti -p 9092:9092 apache/kafka:3.8.0

Python 中的 BinaryIO 類型提示表示一個可以從中讀取字節(jié)并可以寫入字節(jié)的對象。它有 read、write、tell（用于獲取當(dāng)前位置）、seek（用于更改位置）等方法。

現(xiàn)在我們可以實現(xiàn)讀取INT8了：

docker run --rm -ti --net=host apache/kafka:3.8.0 \
  /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

Kafka 使用 big-endian（又名網(wǎng)絡(luò)）字節(jié)排序，因此 byteorder="big"。

現(xiàn)在寫：

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

對于 INT16、INT32 和 INT64，我不會重復(fù)這一點：唯一顯著的區(qū)別是字節(jié)數(shù)（分別為 2、4 和 8）和檢查范圍 ([-(2**15), 2* *15 - 1]、[-(2**31)、2**31 - 1] 和 [-(2**63)、2**63 - 1]相應(yīng)地）。

UINT16 與 INT16 類似：

./gradlew processMessages

注意這里的signed=False。

布爾值

BOOLEAN 本質(zhì)上是帶有額外邏輯的 INT8：== 0 表示 false，!= 0 表示 true。

docker run --rm -ti -p 9092:9092 apache/kafka:3.8.0

您可以在MetadataRequestData生成的類的allowAutoTopicCreation字段中看到BOOLEAN的示例。

浮點數(shù)64

FLOAT64 是雙精度 64 位 IEEE 754 值。 Python 不像 int 那樣有用于 float 的 to_bytes 和 from_bytes 。因此，我們將使用標(biāo)準(zhǔn)庫中的 struct 模塊。

docker run --rm -ti --net=host apache/kafka:3.8.0 \
  /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

>d 表示“大端字節(jié)順序中的雙精度值”。

UNSIGNED_VARINT：可變長度整數(shù)值

可變長度整數(shù)是一種允許在值較小時每個值使用較少位數(shù)的方法。 Kafka 使用 Protocol Buffers 的 Varint 方法。這個想法很簡單：

varint 中的每個字節(jié)都有一個連續(xù)位，指示其后面的字節(jié)是否是 varint 的一部分。這是字節(jié)的最高有效位 (MSB)（有時也稱為符號位）。低7位是有效負(fù)載；生成的整數(shù)是通過將其組成字節(jié)的 7 位有效負(fù)載附加在一起而構(gòu)建的。

詳細(xì)信息可以查看Protobuf規(guī)范和Kafka實現(xiàn)（讀、寫）。

此類型本身不用于協(xié)議字段，但它用于下面描述的緊湊集合。

讓我們實現(xiàn)它。為了信心起見，我們直接從事實來源，Kafka 的 ByteUtils 類中獲取一些示例：

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

運行這個，我們會得到：

def read_exact(buffer: BinaryIO, num_bytes: int) -> bytes:
    value = buffer.read(num_bytes)
    if len(value) != num_bytes:
        raise ValueError(f"Buffer underflow: expected {num_bytes}, got {len(value)}")
    return value

讓我們以可能不是最高效但最簡單的方式來實現(xiàn)它：

def read_int8(buffer: BinaryIO) -> int:
    return int.from_bytes(read_exact(buffer, 1), byteorder="big", signed=True)

通用唯一識別碼

UUID 是用于唯一標(biāo)識實體的 128 位值。例如，它們用于在 CreateTopicsResponse 中傳遞主題 ID。

你可以看到它們在Kafka代碼中是如何讀寫的。重現(xiàn)起來很簡單：

def write_int8(value: int, buffer: BinaryIO) -> None:
    if -(2**7) <= value <= 2**7 - 1:
        buffer.write(value.to_bytes(1, byteorder="big", signed=True))
    else:
        raise ValueError(f"Value {value} is out of range for INT8")

請注意，Kafka 將 null/None 視為零 UUID，因此我們在這里也這樣做。

弦樂

Kafka協(xié)議有4種類型的字符串：

緊湊性指示字符串長度是使用 INT16 還是使用 UNSIGNED_VARINT 編碼。這取決于消息版本（2017年左右推出）?？煽招允侵冈撝凳欠窨梢詾榭?。這也取決于消息的目的和版本（有時字符串字段在協(xié)議演變過程中變得可選）。

字符串在協(xié)議中無處不在。例如，查看生成的類MetadataRequestData.MetadataRequestTopic中的字段名稱。

字符串的編碼非常簡單：首先是長度，然后是 UTF-8 編碼的正文。允許的最大長度為 32767 字節(jié)。空字符串的長度為 -1 并且顯然沒有正文。

由于緊湊型和非緊湊型之間的唯一區(qū)別在于字符串長度的編碼方式，因此我們可以為兩種模式使用一個函數(shù)。

讓我們從讀取和寫入可為空字符串開始：

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

不可為 null 的字符串函數(shù)可以構(gòu)建在這些函數(shù)之上：

./gradlew processMessages

字節(jié)數(shù)組

字節(jié)數(shù)組與字符串非常相似。它們具有相同的潛在可空性和緊湊性：

它們也以相同的方式編碼：長度主體。當(dāng)然，主體不會被視為 UTF-8 字符串，而是被視為不透明的字節(jié)數(shù)組。字節(jié)數(shù)組的最大長度為 2147483647；

您可以在生成的類JoinGroupRequestData.JoinGroupRequestProtocol中找到字段元數(shù)據(jù)中字節(jié)的示例。

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

如您所見，這些函數(shù)與字符串對應(yīng)的函數(shù)之間的差異很小。

其他陣列

該協(xié)議支持字節(jié)以外類型的數(shù)組：字符串、數(shù)字、結(jié)構(gòu)（但不包括嵌套數(shù)組）：ARRAY 和 COMPACT_ARRAY。 緊湊性與字節(jié)數(shù)組和字符串相同。

出于某種原因，協(xié)議規(guī)范中沒有明確提及可空性。但是，數(shù)組可以為空。這是由架構(gòu)定義中的 nullableVersions 控制的，如下所示。

考慮到我們已經(jīng)實現(xiàn)了 read_array_length 和 write_array_length，讓我們實現(xiàn) reader 和 writer 函數(shù)：

./gradlew processMessages

記錄

RECORDS 對 Kafka 記錄進行編碼。該結(jié)構(gòu)非常復(fù)雜，我不會在本指南中描述它（但是，如果您想要它，請在評論中告訴我？?。）為了簡單起見，我們可以將記錄視為 NULLABLE_BYTES 或 COMPACT_NULLABLE_BYTES （取決于消息版本）。

標(biāo)記字段

標(biāo)記字段是 Kafka 協(xié)議的擴展，它允許將可選數(shù)據(jù)附加到消息中。這個想法是雙重的：

1. 如果服務(wù)客戶端不理解標(biāo)記的字段，它會將其保存為未知并忽略它。
2. 如果某個字段很少使用，可以跳過其默認(rèn)值傳輸。

例如，看看這個字段。它有 taggedVersions，它表示從哪個版本開始標(biāo)記該字段（在大多數(shù)情況下，它與添加該字段時的版本相同）。

標(biāo)記字段包含：

1. UNSIGNED_VARINT 類型的標(biāo)簽。
2. COMPACT_BYTES類型的數(shù)據(jù)。

您可以在 KIP-482 中找到有關(guān)標(biāo)記字段的更多詳細(xì)信息。

讓我們實現(xiàn)：

docker run --rm -ti -p 9092:9092 apache/kafka:3.8.0

這里它們的標(biāo)題是“未知”。已知字段需要在其結(jié)構(gòu)內(nèi)進行構(gòu)建。

消息結(jié)構(gòu)

高級消息結(jié)構(gòu)非常簡單。根據(jù)規(guī)范：

docker run --rm -ti --net=host apache/kafka:3.8.0 \
  /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

也就是說，它是一條消息本身，前面有其大?。ㄒ宰止?jié)為單位）。請求和響應(yīng)消息均由緊隨其后的標(biāo)頭組成。由于某種原因，這沒有明確記錄⁴，但你可以相信我嗎？或查看代碼。

請求和響應(yīng)頭

請求頭存在三個版本：0、1、2。它們在協(xié)議中指定為：

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

TAG_BUFFER 就是前面提到的標(biāo)記字段。

讓我們將它們實現(xiàn)為 Python 數(shù)據(jù)類：

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

如您所見，版本 2 中有一些標(biāo)記字段，沒有預(yù)期的已知字段。如果某些標(biāo)記字段被錯誤地發(fā)送到代理，它將被忽略。

響應(yīng)頭存在兩個版本：0和1。它們在協(xié)議中指定為：

./gradlew processMessages

讓我們也實現(xiàn)它們：

docker run --rm -ti -p 9092:9092 apache/kafka:3.8.0

我們沒有實現(xiàn)請求標(biāo)頭的讀取和響應(yīng)標(biāo)頭的寫入。這是為了簡潔起見：在我們的示例中，我們不會發(fā)送響應(yīng)標(biāo)頭并接收請求標(biāo)頭，因為我們不會對服務(wù)器端進行編程。但是，如果您也對服務(wù)器端感興趣，則需要實現(xiàn)這兩個功能（這應(yīng)該很簡單）。

相關(guān)ID

特別注意請求和響應(yīng)標(biāo)頭中的correlation_id 字段。該協(xié)議支持管道：客戶端每個連接可以有多個未完成的請求。相關(guān) ID 允許其將響應(yīng)與請求進行匹配。

標(biāo)頭版本選擇

必須使用哪個版本是API密鑰和消息版本的函數(shù)。目前協(xié)議指南中尚未明確記錄⁵.
參考生成的類ApiMessageType中的requestHeaderVersion和responseHeaderVersion函數(shù)。

發(fā)送請求和接收響應(yīng)

現(xiàn)在，掌握了所有這些知識和代碼，讓我們最終發(fā)送 ApiVersions 請求并接收和讀取響應(yīng)。 ApiVersions 通常是客戶端發(fā)送的第一個請求。其目的是查找代理支持的 API 版本和功能。我們實施了最新版本 3。

在協(xié)議規(guī)范中，定義為：

docker run --rm -ti --net=host apache/kafka:3.8.0 \
  /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

讓我們創(chuàng)建數(shù)據(jù)類：

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

回復(fù)：

def read_exact(buffer: BinaryIO, num_bytes: int) -> bytes:
    value = buffer.read(num_bytes)
    if len(value) != num_bytes:
        raise ValueError(f"Buffer underflow: expected {num_bytes}, got {len(value)}")
    return value

[api_keys] 表示“api_keys 數(shù)組”，其中 api_keys 是下面兩行定義的結(jié)構(gòu)體。

將其轉(zhuǎn)換為 Python 數(shù)據(jù)類：

def read_int8(buffer: BinaryIO) -> int:
    return int.from_bytes(read_exact(buffer, 1), byteorder="big", signed=True)

當(dāng)我們談?wù)摂?shù)組時，我們需要知道我們是否需要緊湊數(shù)組或非緊湊數(shù)組。為了找到答案，我們來看看 ApiVersionsRequest.json 中的架構(gòu)定義?？梢钥吹?"flexibleVersions": "3 "，這意味著從版本 3 開始使用緊湊數(shù)組（更多信息請參見 schema 目錄中的 README.md）。由于我們在這里使用版本 3，因此我們使用緊湊數(shù)組。

實現(xiàn)請求和響應(yīng)類后，我們可以發(fā)送和接收這些請求。對于此 ApiVersions v3，我們需要 v2 請求標(biāo)頭和 v0 響應(yīng)標(biāo)頭（檢查生成的 ApiMessageType.java）。您可以在 ApiVersionsRequest.json 或協(xié)議規(guī)范中找到 API 密鑰 (18)。

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

如果運行此代碼，您將看到控制臺中打印的響應(yīng)標(biāo)頭和消息。恭喜，您已經(jīng)與 Kafka Broker 進行了正確的網(wǎng)絡(luò)交換！

您會注意到 _unknownTaggedFields 中放入了三個標(biāo)記字段。生成的 ApiVersionsResponseData 類的讀寫方法以及 ApiVersionsResponse.json 中的消息定義將幫助您解釋它們?？紤]一下這個作業(yè)嗎？

---

1. 在我的日常工作中，我們開發(fā)了一個開源庫 Kio。它允許我們輕松地從 Python 進行任意 Kafka API 調(diào)用。序列化/反序列化代碼，就像 Kafka 本身一樣，是根據(jù) JSON 協(xié)議定義生成的。生成的代碼經(jīng)過嚴(yán)格測試，包括針對真實 Java Kafka 代碼的屬性測試。??

2. 如果您愿意，也可以使用“消息”：某些模式不適用于 API，但例如對于磁盤上的數(shù)據(jù)。??

3. read_exact 函數(shù)有一個缺點，即當(dāng)?shù)讓泳彌_區(qū)已在內(nèi)存中時，它會復(fù)制數(shù)據(jù)。然而，它對于教育目的來說更方便。??

4. 我做了一個 PR 來解決這個問題。??

5. 我再次發(fā)布了 PR 來解決這個問題。??

以上是Kafka協(xié)議實用指南的詳細(xì)內(nèi)容。更多信息請關(guān)注PHP中文網(wǎng)其他相關(guān)文章！