命令手册

外观

案例：Kafka 消息积压两百万 ​

现象 ​

下游对账系统数据延迟从 1 分钟涨到 4 小时。监控显示消费组 lag 持续增长。

bash
/opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 10.0.0.31:9092 \
  --describe --group order-sync
text
GROUP        TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG      CONSUMER-ID                  HOST
order-sync   order   0          1842000         2284000         442000   consumer-order-sync-1-8f3a  /10.0.0.71
order-sync   order   1          1798000         2248000         450000   consumer-order-sync-2-a1b2  /10.0.0.72
order-sync   order   2          1805000         2255000         450000   consumer-order-sync-3-c3d4  /10.0.0.73
order-sync   order   3          1810000         2260000         450000   consumer-order-sync-4-e5f6  /10.0.0.74

总积压约 179 万，而且还在涨。

定位过程 ​

第一步：消费者数量和分区数是否匹配 ​

bash
/opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 10.0.0.31:9092 \
  --describe --group order-sync --members
text
GROUP        CONSUMER-ID                  HOST        #PARTITIONS
order-sync   consumer-order-sync-1-8f3a /10.0.0.71  1
order-sync   consumer-order-sync-2-a1b2 /10.0.0.72  1
order-sync   consumer-order-sync-3-c3d4 /10.0.0.73  1
order-sync   consumer-order-sync-4-e5f6 /10.0.0.74  1

4 个消费者，topic 只有 4 个分区 —— 并行度已经打满，加实例也没用。

第二步：单条处理耗时多少 ​

应用日志里加埋点后：

text
2026-10-28 03:20:11 INFO  consume order=88213 cost=88ms
2026-10-28 03:20:12 INFO  consume order=88214 cost=92ms
2026-10-28 03:20:13 INFO  consume order=88215 cost=89ms

每条 88~92ms。四台机器、每台一个线程，吞吐约：

text
4 线程 × (1000 / 90) ≈ 44 条/秒

生产端 QPS 约 120，消费只有 44 —— 缺口 76 条/秒，积压必然增长。

第三步：90ms 花在哪 ​

bash
grep -A 3 'consume order=88213' /var/log/order-sync/app.log
java
// 处理逻辑
Order o = parse(record.value());
Account acct = accountClient.get(o.getUserId());      // 同步 HTTP，60ms
BigDecimal rate = rateService.getRate(o.getCurrency()); // 同步 HTTP，20ms
repo.save(o, acct, rate);                              // 5ms

两次同步 HTTP 调用占了 80ms。

第四步：有没有毒丸消息 ​

bash
/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.31:9092 \
  --topic order --partition 0 --offset 1842000 --max-messages 5 \
  --property print.key=true --property print.timestamp=true
text
CreateTime:1761570000000	order-88213	{"id":88213,...}
CreateTime:1761570001000	order-88214	{"id":88214,...}

数据本身没问题，排除毒丸。

处置 ​

立即止血 ​

  1. 加分区 + 加消费者（不要求同 key 严格顺序时可行）：
bash
/opt/kafka/bin/kafka-topics.sh --bootstrap-server 10.0.0.31:9092 \
  --alter --topic order --partitions 24
text
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!
bash
# 扩容消费实例到 12 个（每个 2 线程）
kubectl -n shop scale deploy/order-sync --replicas=12

并行度从 4 提到 24，消费能力约 264 条/秒，可以用 3~4 小时追平积压。

  1. 临时提高批处理大小：
yaml
spring:
  kafka:
    consumer:
      max-poll-records: 200

扩分区会改变 key 的分区归属

如果业务要求同一订单严格顺序，扩分区会导致同一 key 之前和之后的消息落到不同分区，顺序被打乱。

必须扩的话，先确认下游能容忍乱序，或者临时切到"按订单 ID 分区数不变"的临时 topic。

根治 ​

  1. 把同步 HTTP 改成批量/异步：
java
// 批量查
Map<Long, Account> accounts = accountClient.batchGet(userIds);
Map<String, BigDecimal> rates = rateService.batchGet(currencies);

一个 100 条的批次只需要 2 次调用，而不是 200 次。

  1. 本地缓存 + 异步刷新：账户、汇率这类变化不频繁的数据，用 Caffeine 缓存 5 分钟。

  2. 处理与提交解耦：

yaml
spring:
  kafka:
    listener:
      ack-mode: manual_immediate

处理成功后立即提交，避免 rebalance 导致重复。

  1. 毒丸消息进 DLQ：
java
try {
    handle(record);
} catch (Exception e) {
    producer.send(new ProducerRecord<>("order.dlq", record.key(), record.value()));
    log.error("poison message, sent to dlq", e);
}

否则一条坏消息会让整个分区卡住（配合无限重试）。

复盘 ​

问题根因改进
积压 200 万消费 44/s，生产 120/s并行度 + 批量化
加实例无效分区数 = 消费者数扩分区
单条 90ms两次同步 HTTP批量查询 + 缓存
没有积压告警只监控了 broker加 consumer lag 告警
追平慢吞吐上限固定临时扩容 + 峰值预案
bash
# lag 告警（每 5 分钟）
/opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server $B --all-groups --describe \
  | awk 'NR>1 && $6 ~ /^[0-9]+$/ {s[$1]+=$6} END {for (g in s) if (s[g] > 100000) print g, s[g]}'

积压时的标准顺序

  1. 消费者数 < 分区数？→ 直接加实例。
  2. 消费者数 = 分区数？→ 扩分区（注意顺序）。
  3. 单条处理慢？→ 批量化/异步化。
  4. 有坏消息反复重试？→ DLQ。

先看第 1、2 条，成本最低；改造代码是最后手段。