https://blog.csdn.net/qq_45076180/article/details/111561984 --- - Broker: Kafka 集群中的每個服務器稱為一個 Broker，負責存儲和處理消息 - 分區副本（Replication）:* Kafka 通過副本（Replica）機制提供高可用性，每個 Partition 會有多個副本（默認 3 個）。 - 分為 **Leader 副本** 和 **Follower 副本**： * **Leader 副本**：負責讀寫操作。 * **Follower 副本**：被動同步 Leader 的數據，在 Leader 掛掉后，Kafka 自動選舉新的 Leader #### **ISR（In-Sync Replica）** * **ISR**（同步副本集合）：所有與 Leader 副本保持同步的 Follower 副本。 * 只有 ISR 副本才有資格成為新的 Leader > # 消息丟失 - 重試機制: retries 避免因臨時故障丟失數據, enable.idempotence=true + 開啟生產者冪等性 (`enable.idempotence=true`) - Kafka **默認不會每次寫入都立即刷盤（落盤）**，而是**先將數據寫入操作系統的頁面緩存（Page Cache）**，然后**再批量刷入磁盤**, 可以通過修改配置調整入磁盤速度 - 生產者發送消息的時候有個確認機制: - acks=0 : 生產者**不等待** Kafka 確認，直接發送下一條消息 - acks=1 : 只要**Leader 副本**收到消息就返回 ACK，不管 Follower 是否同步 - acks=-1/acks=all : 生產者等待 **Leader 和ISR（同步副本）** 確認消息寫入后才繼續 - 配合 `min.insync.replicas` 指定了**最少需要多少個 ISR 副本**確認消息寫入 - 關閉Unclean 選舉: ``` Broker 1（Leader） ---> 最新數據（消息 1、2、3、4、5） Broker 2（Follower, ISR） ---> 最新數據（消息 1、2、3、4、5） Broker 3（Follower, 非ISR） ---> 舊數據（消息 1、2、3） **正常情況**： * Broker 1 作為 Leader，Broker 2 作為 ISR 復制數據。 * Broker 3 由于同步較慢，被剔除出 ISR（非 ISR） **如果 Broker 1 和 Broker 2 宕機：** * **Unclean 選舉 = `true`** * Kafka 允許 Broker 3 成為新的 Leader，但它**缺少消息 4 和 5**，因此數據丟失。 * **Unclean 選舉 = `false`** * Kafka 不會選出新的 Leader，該分區**不可用**，直到 Broker 1 或 Broker 2 重新上線 ``` - 數據回滾 | Broker | 舊數據（宕機前） | 重新上線后數據 | | --- | --- | --- | | Broker 1 | **1, 2, 3, 4, 5** | **1, 2, 3, 6, 7**（刪除 4、5，改為同步新 Leader） | | Broker 2 | **1, 2, 3, 4, 5** | **1, 2, 3, 6, 7**（刪除 4、5，改為同步新 Leader） | | Broker 3（新 Leader） | **1, 2, 3** | **1, 2, 3, 6, 7** | - 消費者設置 自動提交偏移量: 如果消費失敗, 數據就丟失了(解決方案: 設置手動提交) > # Rebalance 頻繁 - 在 Kafka 中，**Rebalance（再平衡）** 是指 **Kafka Consumer Group（消費者組）重新分配分區（Partition）的過程**。 - 消費者加入或離開 Consumer Group ``` session.timeout.ms=45000 # 默認 10s，適當加大避免誤判 heartbeat.interval.ms=15000 # 默認 3s，適當加大減少心跳頻率 ``` > # Kafka 如何在多個分區中保證消息順序和消息處理效率? - 需要順序消費的, 通過設置key放入同一個分區里 ### **分區分配機制** Kafka 通過**Partitioner**來決定消息發送到哪個分區。默認情況下，Kafka 使用以下規則： 1. **如果有 Key**： * 對 Key 進行哈希（默認使用`murmur2`算法），然后對分區數取模，得到目標分區。 * **公式**：`partition = hash(key) % numPartitions` * **結果**：相同 Key 的消息會被分配到同一個分區。 2. **如果沒有 Key**： * 使用**輪詢策略**（Round Robin）將消息均勻分配到各個分區 --- 消息丟失acks=all + min.insync.replicas ≥ 2 消息重復冪等性 (enable.idempotence=true) + 去重（Redis/數據庫唯一索引） 數據積壓增加消費者 + 批量消費 + 多線程處理 分區不均衡自定義分區策略 + Key 分區 消息順序單個 Partition 保持順序 + StickyAssignor Rebalance 頻繁增加 session.timeout.ms + StickyAssignor 事務問題Kafka 事務 API + 冪等性 性能優化批量發送/消費 + Zero Copy + 壓縮