【車聯網kafka】Kafka核心架構與實戰經驗（第一篇）

一、我與kafka的緣分-初識Kafka

二、Kafka深入探討-了解kafka

?編輯2.1 kafka 生產者框架

2.1.1 生產者在生活中的實例

2.1.2 kafka生產者流程及框架

1.?主線程處理階段

2.?Sender線程處理階段

設計優勢總結

2.2 kafka 生產者框架中的一些關鍵參數

2.3 kafka 生產者框架中一些關鍵問題

2.3.1 kafka如何對消息進行統一處理呢

2.3.2 kafka如何知道發送給哪個分區呢，有哪些分區分配策略？如何發送消息到固定分區？

常見的Kafka生產者分區策略總結

2.3.3 主要用到的序列化方式

2.3.4 如何保證消息發送的一致性、ack如何設置？

第一、ack的設置

第二、確保消息發送

選擇建議與權衡

2.3.5 如何保證消息發送的冪等性

示例配置建議

2.3.6 kafka發送的retry機制

2.3.7 如果發送的消息比較大怎么辦呢

1.?處理超過batch size的消息（默認16KB）

2.?處理超過max.request.size的消息（默認1MB）

3.?處理非常大的消息（如10MB或更大）

整體優化建議總結

OK，今天先到這里后面幾篇會繼續把kafka講完

一、我與kafka的緣分-初識Kafka

? ? ?在車聯網平臺的實際應用中，我們采用 Kafka 作為核心消息中間件實現系統解耦，主要處理車輛 T-Box 設備上報的報文數據。當前平臺已接入 30 萬輛車輛，以每10秒 1 包的頻率持續上傳數據。按此規模計算：

單日數據量：單天的數據量在2.8億條
車輛在線率:? ?30%-40%

選擇 Kafka 主要基于兩個關鍵考量：

海量數據承載能力：其分布式架構與零拷貝機制可穩定支撐 500000 量級/秒的寫入吞吐
業務容錯特性：車輛數據場景允許部分消息丟失（如網絡閃斷），而 Kafka 的異步刷盤機制在保證吞吐的同時，恰能滿足該容忍度

? ? ? 通過 Kafka 的緩沖削峰，我們有效隔離了數據生產端（車輛上行）與消費端（業務處理系統）。此前雖已實際應用，但尚未深入技術原理。現借此機會，我們梳理一下kafka。

二、Kafka深入探討-了解kafka

????????當談到Apache Kafka時，我認為它就是一個高性能的分布式數據流平臺，類似于一個超級大的臨時流轉的存儲倉庫，能夠高效處理海量實時流轉數據。不過，它有兩個核心特點：

????????第一：數據通過生產者"前門進"，通過消費者"后門出"，形成連續的數據管道；

????????第二：由于分布式系統的特性，數據在極端情況下可能會丟失（例如，在故障或配置不當的場景）。

????????因此，對于需要強數據一致性的應用場景（如金融交易或實時記賬系統），Kafka 往往不太合適，因為它更側重于高吞吐量和最終一致性。

2.1 kafka 生產者框架

2.1.1 生產者在生活中的實例

????????想象一下，Kafka的發送者就像是一個送貨員，他的任務很簡單：把貨物（消息）送到指定地方就行。但如果您是負責設計整個送貨系統的“老板”，就不能只盯著送貨員的工作了——您得操心整個鏈條的細節，確保一切高效可靠。這涉及到一系列關鍵問題：

貨物要送給誰？
就像送貨前得知道收貨地址一樣，您需要確定消息具體要發給哪些人或系統。
運輸方式怎么選？
是用小車（高效工具）快速送，還是自己走路（慢速方式）慢慢送？這會影響送貨速度和成本。
貨物需要整理或壓縮嗎？
如果用小車送，是不是得把貨物打包整齊？或者壓縮一下，節省空間，讓一次能送更多貨？
怎么確保貨物送到倉庫？
萬一路上出問題，比如車子壞了，如何保證貨物最終安全到達？需要什么保險措施？
怎么避免一個貨物送兩次？
如何確保每個貨物只送一次，不會重復發送，造成混亂或浪費？
送貨前要檢查貨物嗎？
是否需要先過濾危險品（如有害數據），確保只送安全、有用的貨物？
送貨路徑怎么安排？
是直接送到最終目的地（總部），還是先送到中轉站（配送點），再轉交過去

????????作為系統設計者，您得像一個精明的老板一樣，從全局角度思考這些細節，而不是只關注送貨員本身的動作。那么kafka其實就是一個精明的老板，所有的這些細節都考慮到位了，那么我們可以看一下kafka得“送貨系統”是如何設計的呢？

2.1.2 kafka生產者流程及框架

????????在Kafka生產者中，消息發送過程由兩條線程協同完成：主線程負責消息的預處理和暫存，Sender線程負責消息的發送。這種設計通過批量處理機制（稱為消息批次）顯著提高了吞吐量和效率。以下是詳細流程：

1.?主線程處理階段

消息初始化：主線程生成消息后，首先通過消息攔截器進行轉換（例如，添加元數據或過濾）。
序列化：轉換后的消息必須經過序列化（將對象轉換為字節流），以滿足網絡傳輸要求。
分區選擇：序列化后的消息通過分區器確定目標分區（主題是邏輯概念，實際數據存儲在分區中）。分區器基于鍵或策略選擇合適的分區。
批次積累：消息進入消息累加器等待，與其他消息組合成批次。這種批量機制減少了網絡開銷，提升整體吞吐量。

2.?Sender線程處理階段

批次提取：當批次就緒（例如，達到大小或時間閾值），Sender線程從消息累加器中提取完整批次。
請求封裝：提取的批次被封裝成Request對象，并通過NetworkClient進行排隊和傳輸。
發送執行：NetworkClient將請求發送到Kafka集群，確保可靠性和效率。

設計優勢總結

這種線程分離設計（主線程處理本地邏輯，Sender線程處理網絡I/O）避免了阻塞，并利用批次機制優化資源使用。消息批次是Kafka高吞吐的核心，它通過減少小消息的單獨發送，降低了延遲并提升了集群性能。

2.2 kafka 生產者框架中的一些關鍵參數

參數名稱	描述
key.serializer 和 value.serializer	指定發送消息的 key 和 value 的序列化類型。必須使用全類名。
buffer.memory	RecordAccumulator 緩沖區總大小，默認值為 32m。
batch.size	緩沖區一批數據最大值，默認值為 16k。適當增加該值可提高吞吐量，但設置過大可能導致數據傳輸延遲增加。
linger.ms	如果數據未達到 batch.size，sender 在等待該時間后發送數據。單位為 ms，默認值 0ms（無延遲）。生產環境建議設置為 5-100ms。
acks	應答機制：0-生產者發送數據后不需落盤應答；1-Leader 收到數據后應答；-1(all)-Leader 和 isr 隊列所有節點收齊數據后應答。默認值 -1，等價于 all。
max.in.flight.requests.per.connection	允許最多未返回 ack 的次數，默認為 5。開啟冪等性時，該值需在 1-5 范圍內。
retries	消息發送錯誤時系統重發次數，默認值為 int 最大值（2147483647）。若需保證消息有序性，需設置 max.in.flight.requests.per.connection=1，否則重試失敗消息時其他消息可能已發送成功。
retry.backoff.ms	兩次重試之間的時間間隔，默認值為 100ms。
enable.idempotence	是否開啟冪等性，默認值為 true（開啟）。
compression.type	生產者發送數據的壓縮方式，默認值為 none（不壓縮）。支持類型：none、gzip、snappy、lz4 和 zstd。

2.3 kafka 生產者框架中一些關鍵問題

2.3.1 kafka如何對消息進行統一處理呢

通過 kafka 生產者的 攔截器進行消息的轉化處理。一般我不會使用，畢竟消息都是處理好發送的，沒必要在攔截一層在進行處理。

????????KafkaProducer在將消息序列化和計算分區之前會調用生產者攔截器的onSend（）方法來對消息進行相應的定制化操作。一般來說最好不要修改消息 ProducerRecord 的 topic、key 和partition 等信息，如果要修改，則需確保對其有準確的判斷，否則會與預想的效果出現偏差。比如修改key不僅會影響分區的計算，同樣會影響broker端日志壓縮（Log Compaction）的功能。

????????KafkaProducer 會在消息被應答（Acknowledgement）之前或消息發送失敗時調用生產者攔截器的 onAcknowledgement（）方法，優先于用戶設定的 Callback 之前執行。這個方法運行在Producer 的 I/O 線程中，所以這個方法中實現的代碼邏輯越簡單越好，否則會影響消息的發送速度。

????????close（）方法主要用于在關閉攔截器時執行一些資源的清理工作。

2.3.2 kafka如何知道發送給哪個分區呢，有哪些分區分配策略？如何發送消息到固定分區？

????????在Kafka中，組裝ProducerRecord消息時，可以通過判斷消息的key或value來顯式指定分區號。這種方式在技術上可行，但不推薦，因為它會將分區選擇邏輯與消息組裝過程耦合，導致代碼可維護性降低。例如，業務邏輯變更可能直接影響分區策略，增加系統復雜度。

1. 未指定分區號時的默認行為，如果未顯式指定分區號（例如：ProducerRecord record = new ProducerRecord("test-topic", "key-" + i, s);），Kafka會使用默認的DefaultPartitioner分區器：

2.如果我們在進行消息組裝的時候，指定了分區號（例如：ProducerRecord record = new ProducerRecord("test-topic",1，"key-" + i,s)，那么，kafka就不會使用分區器。

常見的Kafka生產者分區策略總結

默認分區策略（Default Partition Strategy）
- 基于消息的key決定分區：如果key不為null，則使用key的哈希值（例如，通過哈希函數計算分區索引）；如果key為null，則自動切換到輪詢策略。
- 優點：簡單高效，適用于大多數場景，能保證相同key的消息分配到同一分區以實現有序性。
- 缺點：當key分布不均勻時，可能導致分區負載不均衡。
輪詢策略（Round-Robin Partition Strategy）
- 當消息key為null時，生產者自動使用此策略：消息依次循環發送到所有可用分區（例如，分區0、1、2...然后重復）。
- 優點：確保消息均勻分布，避免單個分區過載；適用于無key或低順序要求的場景。
- 缺點：不保證相同key的消息順序性，可能影響某些應用的一致性需求。
自定義分區器（Custom Partitioner）
- 用戶可通過實現Partitioner接口（如Java中的org.apache.kafka.clients.producer.Partitioner）自定義邏輯，基于消息key、value或其他屬性（如時間戳或業務ID）計算分區。
- 優點：高度靈活，能適配復雜需求（如根據地理區域或用戶ID分區）；支持集成外部系統。
- 缺點：需要額外開發，可能引入性能開銷；需確保邏輯正確以避免分區熱點。
一致性哈希分區（Consistent Hashing）
- 非Kafka原生支持，但可通過自定義分區器實現：使用一致性哈希算法（例如，基于虛擬節點）在多個維度（如多個key或屬性）上均勻分布數據，減少分區重平衡時的數據遷移。
- 優點：提升擴展性和容錯性，特別適合動態集群環境；能平衡負載并減少“熱點”問題。
- 缺點：實現復雜，需自定義開發；哈希計算可能增加延遲。

2.3.3 主要用到的序列化方式

????????我們公司這邊對key和value都采用String序列化方式。對于復雜的對象，我們一般使用json轉化包將我們復雜的對象轉化成 json字符串，然后就可以使用我們的String的序列化方式了

2.3.4 如何保證消息發送的一致性、ack如何設置？

第一、ack的設置

????????Kafka的acks參數是生產者端的配置，用于控制消息副本的寫入確認級別。它影響消息的持久性和系統吞吐量，具體分為三個級別：

acks = 0：生產者不等待任何服務器確認。這提供了最高吞吐量（延遲最低），但安全性最低。如果服務器故障，消息可能丟失。適用場景包括：
- 網站點擊量統計
- 頁面停留時間記錄
- 視頻播放量追蹤（這些場景容忍少量數據丟失）
acks = 1：默認設置。生產者等待首領節點（Leader）確認寫入成功即可。相比acks=0，吞吐量略有下降，但可靠性提高（首領節點故障時可能丟失消息）。適合大多數實時數據處理，如用戶行為分析。
acks = all（或acks = -1）：生產者等待所有副本節點（包括ISR列表）都確認寫入成功。這提供最高可靠性（消息幾乎不會丟失），但吞吐量最低、延遲最高。適用場景包括：
- 金融交易記錄
- 關鍵事件審計（如支付確認）

第二、確保消息發送

三種消息發送確認機制的優劣比較：

機制類型	吞吐量	可靠性	實時反饋	資源消耗	適用場景
消息攔截器（onAcknowledgement方法）	低	高	實時	高	需要每條消息精確追蹤的場景
同步發送(Future.get)	低	高	實時	中	強一致性要求的低頻場景
異步回調(Callback)	高	高	異步	低	高吞吐量要求的業務場景

采用第三種方式往往可以具備高吞吐，推薦?

import org.apache.kafka.clients.producer.*;
import java.util.Properties;public class KafkaProducerOptimized {public static void main(String[] args) {// 1. 配置Producer屬性Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("batch.size", 16384); // 批量大小，單位字節，優化吞吐量props.put("linger.ms", 5); // 發送延遲，允許批量累積，提高吞吐// 2. 創建Producer實例try (Producer<String, String> producer = new KafkaProducer<>(props)) {// 3. 發送消息，并添加Callbackfor (int i = 0; i < 10; i++) { // 示例：發送多條消息ProducerRecord<String, String> record = new ProducerRecord<>("your-topic", "key-" + i, "value-" + i);producer.send(record, new Callback() {@Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if (exception instanceof RetriableException) {// 可重試異常} else {// 不可重試異常，記錄并放棄}
}}});}} catch (Exception e) {// 全局異常處理System.err.println("Producer異常: " + e.getMessage());}}
}

選擇建議與權衡

沒有固定“最佳”設置，需根據業務需求平衡性能和安全性：

性能優先：選擇acks=0，適用于高吞吐、低延遲場景（如監控數據）。
可靠性優先：選擇acks=all（Kafka），適用于數據一致性要求高的系統（如庫存管理）。
一般場景：acks=1（Kafka）是良好折衷，兼顧效率和基本可靠性。

2.3.5 如何保證消息發送的冪等性

????????在分布式系統中，使用 Apache Kafka 進行消息傳遞時，保證消息發送的冪等性至關重要。冪等性確保無論操作執行多少次（例如，由于網絡重試或故障恢復），消息處理結果都不會改變，從而避免數據重復和不一致。Kafka 通過以下核心機制實現這一目標：

生產者配置：
- 啟用冪等性：設置?enable.idempotence=true，生產者會為每個消息分配一個唯一的 PID（生產者 ID）和一個單調遞增的序列號（例如，序列號從 0 開始遞增）。這樣，Kafka broker 可以識別并丟棄重復消息。
- 確認模式：推薦設置?acks=all，確保生產者收到所有副本的確認，提升可靠性。這與冪等性結合使用，能有效處理網絡問題導致的重發。
序列號和 PID 管理：
- 每個生產者實例分配一個唯一 PID，每個消息附帶一個序列號。Kafka broker 基于 PID 和序列號檢查消息是否重復，如果序列號不連續或重復，則拒絕處理。
存儲機制：
- Kafka 的日志存儲設計（如分區日志文件）持久化消息狀態。通過記錄序列號和 PID，即使在 broker 故障恢復后，也能從日志中恢復并避免重復處理。
事務支持：
- 事務本身不直接提供冪等性，但可與冪等性結合使用（例如，設置?enable.idempotence=true?并開啟事務）。這確保了跨分區操作的原子性和一致性，適用于復雜場景（如多消息事務提交）。
客戶端庫支持：
- 使用最新 Kafka 客戶端庫（如 Kafka Java 客戶端），確保實現無缺陷。早期版本可能不支持某些功能，因此升級庫版本是必要的最佳實踐。

示例配置建議

enable.idempotence=true  # 啟用冪等性
acks=all                 # 確保所有副本確認
retries=2147483647       # 設置高重試次數（接近最大值），配合冪等性處理網絡重試

注意事項：冪等性僅針對單個生產者實例的單個分區有效。如果涉及多分區或復雜事務，需額外配置事務管理。實際部署中，測試重試場景以驗證配置可靠性。

2.3.6 kafka發送的retry機制

? ? ? ? 一般情況下我們這邊沒有使用過kafka自己的retry機制，我們通過kafka的callback機制對發送失敗的消息進行監聽，然后再次進行發送（失敗的消息保存到db中，日志中）

? ? ? ? 如果使用retry可以通過retries 參數控制重試的次數，通過retry.backoff.ms控制重試次數之間的時間間隔。間隔是500ms，之所以設置為500ms，kafka中topic分區的副本首領選舉的整個過程是500ms以內完成的。

????????kafka的retry這機制，不是每次都retry的，如果收到了這樣的error：

????????消息大小超過了request.max.size 或者超過了message.max.bytes 類似這樣的錯誤，kafka是不會選擇重試的，因為這種錯誤是無法通過重試而成功的。

????????如果是因為網絡延遲、網絡抖動啊、分區的一系列暫時不可用啊，這種錯誤kafka認為有可能在重試的過程中成功。

2.3.7 如果發送的消息比較大怎么辦呢

? ? ? ? 我們這邊消息在1MB以下，Kafka的設計初衷是處理高吞吐量的小消息，對于大消息（如超過16KB或1MB）存在性能瓶頸，需要針對性調整參數或采用替代方案。以下是分步優化建議：

1.?處理超過batch size的消息（默認16KB）

Kafka的batch size默認為16KB。如果單個消息超過16KB，消息累加器會直接發送該消息，而不是將其拆分成多個batch組裝。這避免了消息分割的開銷，但引入了內存性能問題：
- 對于小于或等于16KB的消息，Kafka復用java.io.ByteBuffer（一個固定大小的緩沖區），減少了內存申請和GC壓力。
- 對于大于16KB的消息，Kafka無法復用ByteBuffer，每次都需要申請新內存空間（大小等于消息本身）。這會導致頻繁的內存分配和垃圾回收，影響性能（如吞吐量下降、延遲增加）。
優化建議：適當調大batch size（例如調整為32KB或64KB），通過生產者配置參數batch.size設置。這可以增加消息復用的概率，減少內存開辟開銷。但需注意，batch size過大可能增加延遲（等待更多消息填充batch），需根據業務場景平衡。

2.?處理超過max.request.size的消息（默認1MB）

Kafka生產者參數max.request.size限制單條消息最大大小（默認1MB）。如果消息超過此限制（如2MB），僅調整該參數不夠，還需協調broker和消費者配置：
- 生產者端：將max.request.size調整為2MB（或更大），確保生產者能發送大消息。
- Broker端：Broker通過message.max.bytes參數控制接收的最大消息大小（默認1MB）。需將其至少調整為2MB，以匹配生產者設置。否則，broker會拒絕過大消息。
- 消費者端：消費者通過fetch.max.bytes參數控制每次拉取的最大消息大小（默認1MB）。需將其至少調整為2MB，確保消費者能接收和處理大消息。
- 經驗策略：設置參數時，遵循max.request.size?<?message.max.bytes?<?fetch.max.bytes（例如，生產者2MB、broker 2.1MB、消費者2.2MB）。這避免因參數不一致導致消息拒絕或失敗。調整后，重啟Kafka集群生效。

3.?處理非常大的消息（如10MB或更大）

Kafka不適合處理超大消息（如10MB以上），因為：
- 內存開銷大：每次發送都需要申請新空間，增加GC壓力。
- 網絡和存儲瓶頸：大消息傳輸慢，可能阻塞broker線程，影響整體吞吐量。
- 設計局限：Kafka優化于小消息流式處理，大消息會破壞其性能優勢。
替代方案：優先考慮非Kafka組件，如：
- 使用文件傳輸協議（如SFTP、HTTP）或網絡附加存儲（NAS）傳輸大文件。
- 將大消息存儲在對象存儲（如S3），然后通過Kafka發送文件引用（如URL）。
如果必須使用Kafka的優化策略：
- 策略1：消息拆分與順序保證
  - 將大消息拆分成多個小消息（每個小于batch size或max.request.size）。
  - 生產者指定所有小消息發送到同一分區（使用相同key），并采用單線程發送，確保消息順序（如先發消息頭，再發消息體）。
  - 消費者端按順序接收并組裝小消息還原為大消息。這需要應用層邏輯支持，但能避免Kafka大消息的性能問題。
- 策略2：消息壓縮
  - 啟用生產者壓縮，通過參數compression.type設置壓縮算法（如snappy、gzip或lz4）。壓縮可減少消息大小（例如，gzip壓縮率可達70%），降低網絡和內存開銷。
  - 示例：設置compression.type=gzip，生產者自動壓縮消息，消費者自動解壓。但需注意，壓縮會增加CPU開銷，需測試性能影響。

整體優化建議總結

預防為主：在業務設計階段，避免發送大消息（如超過1MB）。優先拆分或壓縮消息。
參數調優：
- 調大batch.size（如32KB）以優化內存復用。
- 對大消息（>1MB），協調調整max.request.size、message.max.bytes和fetch.max.bytes，確保三者匹配且略遞增。
- 監控Kafka性能（如使用JMX或Kafka監控工具），根據指標調整參數。
性能權衡：調大參數可能增加內存使用和延遲，測試環境驗證后再上線。
大消息處理原則：對于>10MB消息，強烈建議使用替代方案。如果必須用Kafka，結合拆分和壓縮策略。