Kafka 實務注意事項：廣播怎麼做、順序怎麼保、group.id 怎麼取，以及 K8s 自動擴展為什麼沒有變快

Kafka 的基本元件可以先參考 Kafka 完整入門。這篇只講一件事背後的各種變化：group.id 與 Partition 的分配規則。幾乎所有 Kafka 在應用上的坑，最後都會回到這兩條規則：

1. 同一個 Group 內，一個 Partition 只會分給一個 Consumer（分工）
2. 不同的 Group 之間互不影響，各自讀到完整的資料（各讀一份）

一、廣播：讓每個人都收到同一則訊息

Kafka 沒有像 RabbitMQ Fanout Exchange 那樣「廣播」的設定，廣播是用 group.id 做出來的。依「誰要收到」分成兩種：

不同的服務都要收到：各自用不同的 group.id

訂單建立後，同步服務要寫進搜尋引擎、通知服務要發推播、稽核服務要留紀錄。三個服務讀同一個 Topic，各自使用自己的 group.id：

order-sync      → 3 個 Pod 分工處理
order-notify    → 2 個 Pod 分工處理
order-audit     → 1 個 Pod

每個服務都讀到完整的訊息，服務內的多個 Pod 再彼此分工。Producer 完全不用知道有誰在消費，之後多一個服務，只要用新的 group.id 訂閱就好。

同一個服務的每個 Pod 都要收到：每個 Pod 一個 group.id

有些事情是每個實例都要做的，例如：

• 物件資料更新，每個 Web Pod 的本機記憶體快取都要清掉
• 使用者透過 WebSocket 連在不同的 Pod 上，推播訊息要讓每個 Pod 都收到，再由連著那個使用者的 Pod 送出去

這時候如果全部 Pod 共用一個 group.id，一則訊息只會被其中一個 Pod 收到，其他 Pod 的快取就永遠是舊的。做法是讓每個 Pod 有自己的 group.id：

var config = new ConsumerConfig
{
    BootstrapServers = "kafka:9092",
    // 在 K8s 裡 MachineName 就是 Pod 名稱，每個 Pod 自成一個 Group
    GroupId = $"cache-invalidate-{Environment.MachineName}",
    // 新 Pod 只需要處理之後的事件，不用從頭讀
    AutoOffsetReset = AutoOffsetReset.Latest
};

要注意的地方：

• 每次部署都會產生新的 Group：Pod 名稱每次都不同，舊的 Group 雖然沒人用了，但它的 Offset 紀錄會留在 Kafka，直到 offsets.retention.minutes（預設 7 天）後才清除。功能上沒有影響，但監控 Consumer Lag 的儀表板會出現一堆停止消費、Lag 一直增加的 Group，告警規則要排除這個前綴，不然會一直誤報
• 另一種做法是不加入 Group：直接用 Assign 指定讀取所有 Partition、從尾端開始讀、不提交 Offset。沒有 Group 就沒有 Rebalance，也不會留下多餘的 Offset 紀錄，但新增 Partition 時要自己處理
• 這種用法本質上是「收到就做、漏了也沒關係」：Pod 重啟的那幾秒內發出的訊息它收不到。如果漏掉會出問題，快取一定要再加上 TTL 當作保險

二、按順序推送：Key 只是第一步

順序是 Kafka 最常被誤解的地方。「用帳號 ID 當 Key」只保證了同一個帳號的訊息進同一個 Partition，但從 Producer 到 Consumer 處理完，中間還有好幾個地方會把順序打亂。

先決定需要哪一種順序

真正需要「全域順序」的情況其實很少，大部分需求仔細想都是「同一個帳號、同一筆訂單內要有序」。

Producer 端：重試可能讓順序對調

Producer 會同時送出多個批次（max.in.flight.requests.per.connection，預設 5）。如果批次 1 失敗重試、批次 2 先成功了，寫進 Partition 的順序就變成 2、1。

• 開啟冪等（enable.idempotence=true）：Broker 會依序號檢查，在 in-flight 不超過 5 的情況下仍然保證順序。Java Client 預設開啟，.NET 的 Confluent.Kafka 預設是關的，要自己打開
• 如果因為某些原因不能開冪等，就要把 max.in.flight.requests.per.connection 設成 1，代價是吞吐量下降
• 應用程式自己也不能亂：同一個帳號的事件如果是從多個執行緒同時呼叫 Produce，送出的順序本身就不確定，Kafka 也救不了

Consumer 端：最常見的錯誤是丟進執行緒池

Partition 數量有限，單一 Consumer 處理太慢時，很自然會想「拉回來之後丟給 Task.Run 平行處理」。但這樣一來，同一個帳號的「下單」和「取消」被兩個執行緒同時處理，誰先完成完全看運氣，前面所有保證順序的努力全部白費。

需要在 Consumer 內部加速時，要依 Key 分流：

// 8 條處理通道，同一個 Key 永遠進同一條，通道內依序處理
var lanes = Enumerable.Range(0, 8)
    .Select(_ => Channel.CreateBounded<ConsumeResult<string, string>>(100))
    .ToArray();

while (!ct.IsCancellationRequested)
{
    var result = consumer.Consume(ct);
    var lane = (int)((uint)result.Message.Key.GetHashCode() % (uint)lanes.Length);
    await lanes[lane].Writer.WriteAsync(result, ct);
}

這樣同一個帳號依序處理、不同帳號平行處理。但要注意 Offset 不能隨便提交：通道 1 處理到 Offset 105、通道 2 才處理到 98，這時只能提交 98，否則當掉重啟後 99～104 就漏掉了。這部分容易寫錯，先確認增加 Partition 與 Consumer 能不能解決，再考慮自己做。

失敗重試會卡住後面所有訊息

有順序的 Partition 裡，一則訊息處理失敗時只有兩個選擇：

• 原地重試：後面的訊息全部等它，這個 Partition 停住，Lag 一直上升（Head-of-line Blocking）
• 丟到另一個重試 Topic、先處理後面的：Partition 不會卡住，但這個帳號的順序就被打亂了

沒有兩全其美的做法。交易類的事件通常選擇原地重試、加上退避時間，並且一定要對 Lag 告警，讓工程師盡快介入；如果業務允許，也可以只把「同一個 Key 後續的訊息」一起暫存起來，其他 Key 繼續處理，但實作複雜度高很多。

增加 Partition 會打亂 Key 的對應

hash(key) % N 的 N 改變後，同一個帳號會被分到不同的 Partition。擴充的那一刻，新 Partition 上的新訊息可能比舊 Partition 上還沒處理完的舊訊息先被處理。順序敏感的 Topic 一開始就要開足 Partition，非擴充不可時，要先停止寫入、等 Consumer 把 Lag 消化完再擴。

三、相同 group.id 會發生什麼事

group.id 只是一個字串，Kafka 不會檢查「這兩個 Consumer 是不是同一個服務」。只要字串一樣，就會被當成同一個 Group 一起分工。這帶來幾個常見事故：

兩個不同的服務用了同一個 group.id

通知服務是從同步服務複製過來改的，設定檔裡的 group.id 忘了改。兩個服務都叫 order-sync：

• 6 個 Partition 被分給兩個服務的所有 Pod，同步服務只拿到一部分 Partition，通知服務拿到另一部分
• 結果就是搜尋引擎少了一部分訂單、一部分使用者沒收到通知，而且沒有任何錯誤訊息，兩邊的 Log 看起來都正常

這和 RabbitMQ 中兩個服務共用同一個 Queue 是完全一樣的問題（參考 RabbitMQ 的 Exchange、Queue、Binding）。

本機開發連到正式環境的 Kafka

工程師在自己的電腦上 Debug，appsettings.Development.json 不小心指到了正式環境的 Kafka，group.id 又和正式環境一樣。這台筆電就加入了正式環境的 Group，分走了幾個 Partition：正式環境的部分訂單被筆電上的程式「處理」掉了，Offset 也被提交，正式環境的 Pod 再也不會處理這些訊息。

同一個 group.id 訂閱不同的 Topic

服務 A 用 group-x 訂閱 Topic 1、服務 B 也用 group-x 訂閱 Topic 2。雖然讀的資料不重疊，但它們是同一個 Group：服務 A 部署、重啟時觸發的 Rebalance，也會讓服務 B 一起暫停。

怎麼避免

• group.id 命名固定格式：{服務名稱}-{用途}，例如 order-sync-es、order-notify-push，一個服務的一個用途就是一個 Group
• group.id 不放在可以隨意複製的共用設定檔裡，而是和服務名稱綁在一起產生
• 正式環境的 Kafka 開啟驗證與 ACL，開發環境的帳號根本沒有權限加入正式環境的 Group
• 定期用 kafka-consumer-groups.sh --describe 檢查每個 Group 的成員，看到不認識的 Client ID 或主機名稱就要追查

四、K8s 自動擴展：Pod 變多了，為什麼沒有變快？

這是一個很容易發生的情境：

• Topic order-events 有 6 個 Partition
• Consumer 服務部署在 K8s，設定了 HPA：CPU 使用率 70%、最少 3 個、最多 20 個 Pod
• 某天流量暴增，Lag 一路上升，HPA 也如預期把 Pod 擴到了 12 個

但 Lag 完全沒有下降得比較快。原因就是第一條規則：一個 Partition 只能分給 Group 內的一個 Consumer。6 個 Partition 最多只有 6 個 Pod 在工作，另外 6 個 Pod 只是佔著資源、什麼都沒做。

從監控上看起來是「有 12 個 Pod 在扛流量，很 HA」，實際上一半的 Pod 是閒置的。而且情況還會更糟：

• 閒置的 Pod 把平均 CPU 拉低：6 個工作中的 Pod 都滿載，加上 6 個閒置的，平均 CPU 只有一半，HPA 判斷「不需要這麼多」又開始縮減
• 每次擴展、縮減都會觸發 Rebalance：Pod 加入或離開 Group，Partition 就要重新分配，分配期間相關的 Partition 停止消費。HPA 來回擴縮，Group 就一直在 Rebalance，實際消費的時間反而變少
• CPU 可能根本不是瓶頸：如果 Consumer 慢是因為寫資料庫很慢，Pod 都在等 I/O，CPU 很低，HPA 完全不會擴展，Lag 卻一直累積

正確的做法

1. 最大副本數不要超過 Partition 數

這是最基本的一條。想要更多 Consumer 平行處理，要先增加 Partition（並考慮前面講的順序問題），而不是只調 HPA 的上限。

2. 依 Consumer Lag 擴展，而不是 CPU

Consumer 該不該擴展，真正的指標是「還有多少訊息沒處理」。用 KEDA 的 Kafka Scaler，可以直接依 Group 的 Lag 擴展：

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: order-consumer
spec:
  scaleTargetRef:
    name: order-consumer          # 對應的 Deployment
  minReplicaCount: 2
  maxReplicaCount: 6              # 不超過 Partition 數
  cooldownPeriod: 300             # Lag 消化完後等 5 分鐘再縮，避免來回擴縮
  triggers:
    - type: kafka
      metadata:
        bootstrapServers: kafka:9092
        consumerGroup: order-processor
        topic: order-events
        lagThreshold: "500"       # 平均每個 Pod 負責 500 筆 Lag

KEDA 的 Kafka Scaler 有一個 allowIdleConsumers 設定，預設是 false，代表算出來的副本數不會超過 Partition 數，即使 maxReplicaCount 設得比較大也一樣，等於內建了第一條規則。

3. 減少 Rebalance 的衝擊

• 使用漸進式 Rebalance：partition.assignment.strategy=cooperative-sticky，Pod 加入或離開時只搬動需要移動的 Partition，其他 Pod 照常消費
• 優雅關閉：K8s 送出 SIGTERM 時，Consumer 要主動呼叫 Close() 離開 Group 並提交 Offset。否則 Group 要等 session.timeout.ms（預設 45 秒）才發現這個 Pod 不見了，這段時間它負責的 Partition 完全沒人處理。terminationGracePeriodSeconds 也要大於處理一批訊息的時間
• Static Membership：設定 group.instance.id（.NET 是 GroupInstanceId）給每個 Consumer 一個固定身分，搭配 StatefulSet 固定的 Pod 名稱。Pod 重啟只要在 session.timeout.ms 內回來，就拿回原本的 Partition，完全不會觸發 Rebalance，很適合滾動更新頻繁的服務

protected override Task ExecuteAsync(CancellationToken stoppingToken) => Task.Run(() =>
{
    _consumer.Subscribe("order-events");
    try
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            var result = _consumer.Consume(stoppingToken);
            Handle(result);
            _consumer.Commit(result);
        }
    }
    catch (OperationCanceledException) { }
    finally
    {
        // 收到 SIGTERM 時主動離開 Group，讓 Partition 立刻交給其他 Pod，而不是等 45 秒逾時
        // Consumer 不是執行緒安全的，Close 要在同一個執行緒呼叫
        _consumer.Close();
    }
}, stoppingToken);

那 HA 呢？

多開幾個閒置的 Pod 當「備援」，其實沒有想像中有用：Kafka 的容錯本來就是透過 Rebalance，任何一個還活著的 Consumer 都能接手掛掉的那個的 Partition，不需要專門待命的 Pod。

Consumer 的高可用要做的是：

• 至少 2 個以上的副本，而且用 topologySpreadConstraints 或 Pod Anti-Affinity 分散在不同的 Node，避免一台 Node 掛掉就全部消失
• 副本數不超過 Partition 數，每個 Pod 都在工作
• 做好優雅關閉，讓計畫中的重啟能立刻交接，而不是等逾時

一句話總結

廣播靠的是不同的 group.id，服務之間各用一個，需要每個 Pod 都收到就每個 Pod 一個；順序要從 Key、Producer 的冪等、Consumer 不丟進執行緒池一路顧到底；group.id 一樣就會被當成同一組分工，命名與權限都要管好；而 Consumer 在 K8s 上擴展，副本數超過 Partition 數只會多出閒置的 Pod 和更多的 Rebalance，要依 Lag 擴展、上限對齊 Partition 數，才是真的有在做事的擴展。