Kafka 消费积压治理实战：rebalance 风暴、分区键顺序与批量幂等

分类：数据库与中间件 · 作者：管理员 · 发布于 2026-10-27 · 阅读 0 · 标签：Kafka / 消息队列 / Spring Boot / 幂等 / 分布式

一、故障现场：一夜之间 lag 涨到 80 万

订单事件流是电商中台里最典型的异步链路：订单服务把「已支付」「已退款」「已发货」写进同一个 topic，下游库存、积分、履约三个消费组各自订阅。这条链路平时很安静，直到某次大促当晚，监控上库存消费组的 lag 从几百一路爬到 80 万，同时日志里开始出现下面这行：

org.apache.kafka.clients.consumer.CommitFailedException:
Offset commit cannot be completed since the consumer is not part of
an active group and therefore cannot commit. This means that either the
consumer is not active, or it has left the group after a rebalance.

这句话几乎是 Kafka 消费端最容易被误读的报错。很多人第一反应是「消费者挂了」，实际上它表达的是：消费者还在跑，但它在提交位移之前已经被踢出了消费组。踢出动作由 broker 端的 group coordinator 发起，触发条件只有一个——超过 max.poll.interval.ms 没有调用 poll()。

把这条因果链摊开就很清楚了：单条消费内部串行调用了扣库存、加积分、写履约单三个远程接口，平均 600ms，遇到慢 SQL 或下游抖动直接飙到 30s；默认 max.poll.interval.ms=300000（5 分钟）看着很宽裕，但 200 条一批的拉取量乘以 600ms，一批就要 120 秒，再叠加两次超时就直接越过阈值。于是 rebalance 发生，分区被重新分配，未提交的位移回退，同一批消息被另一个消费者重新消费——lag 不但没降，反而因为重复消费越滚越大。

二、先分清两类问题：吞吐不够，还是语义错误

处理积压之前必须先判断性质，否则很容易用错药。

吞吐不足：lag 线性上涨，消费速率稳定但低于生产速率，没有 rebalance 日志。解法是提高并行度、批量处理、去掉消费链路里的同步远程调用。
语义错误：出现 rebalance、重复消费、消费卡死在某条毒消息上。此时提高并发度只会让重复消费更严重，必须先把「提交语义 + 幂等」补上。

本次故障两者叠加：既有慢消费导致的 rebalance，也有下游扣减不幂等导致的重放放大。所以改造顺序是先保证语义正确，再谈吞吐。

三、消费端改造：批量拉取 + 手动提交 + 幂等表

第一步把消费模型从「一条一个远程调用」改成「一批一次批量落库」，这能把 200 次网络往返压缩成 1 次 SQL。配置侧的关键项如下（broker 地址一律用占位符）：

spring:
  kafka:
    consumer:
      bootstrap-servers: broker-a:9092,broker-b:9092
      group-id: stock-consumer
      enable-auto-commit: false        # 关掉自动提交，改成显式 ack
      max-poll-records: 200            # 一批最多 200 条
      fetch-max-wait: 200ms            # 有数据别等太久，降低端到端延迟
      max-poll-interval-ms: 300000
      properties:
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor
    listener:
      type: batch
      ack-mode: manual_immediate
      concurrency: 3

这里有两个容易忽略的点：CooperativeStickyAssignor 是增量式再均衡协议，rebalance 时只迁移需要移动的分区，不会像 eager 协议那样让整个消费组停摆重来；concurrency 的可调上限等于订阅 topic 的分区数，设成 8 但只有 3 个分区，多出来的 5 个线程全程空转，只是白白占内存。

消费方法本身：

@KafkaListener(topics = "order-events", groupId = "stock-consumer")
public void onBatch(List<ConsumerRecord<String, String>> records,
                    Acknowledgment ack) {
    List<OrderEvent> events = new ArrayList<>(records.size());
    for (ConsumerRecord<String, String> record : records) {
        OrderEvent event = EventCodec.decode(record.value());
        if (event != null) {
            events.add(event);
        }
    }

    // 1) 幂等过滤：event_id 唯一约束，重复的直接跳过
    List<OrderEvent> fresh = idempotentRepo.filterAndMark(events);
    if (fresh.isEmpty()) {
        ack.acknowledge();
        return;
    }

    // 2) 同一批内按 skuId 合并，减少行锁争抢
    Map<Long, Integer> delta = new HashMap<>();
    for (OrderEvent event : fresh) {
        delta.merge(event.getSkuId(), event.getQty(), Integer::sum);
    }

    // 3) 一次批量扣减，扣减条件里仍保留 stock >= qty 兜底
    stockRepository.batchDeduct(delta);

    // 4) 全部成功后才提交位移
    ack.acknowledge();
}

幂等表是整套方案的基石，靠数据库唯一约束而不是应用层「先查后插」，后者在并发下必然漏判：

CREATE TABLE consumed_event (
  event_id    varchar(64) PRIMARY KEY,
  consumer    varchar(64) NOT NULL,
  consumed_at timestamptz NOT NULL DEFAULT now()
);

-- 批量写入，冲突即为已消费，DO NOTHING 不会抛异常打断整批
INSERT INTO consumed_event(event_id, consumer) VALUES (?, ?)
ON CONFLICT (event_id) DO NOTHING;

注意 ON CONFLICT DO NOTHING 拿不到「哪几条真的插入了」。要精确过滤，用 INSERT ... ON CONFLICT DO NOTHING RETURNING event_id 收集返回集，或者先批量 SELECT event_id FROM consumed_event WHERE event_id = ANY(?) 得到已存在集合再取差集——后者更直观，且能顺手把过滤结果打进日志，便于事后核对。

四、顺序性：分区键选错，比不顺序更糟

订单事件天然要求「同一订单的支付、退款、发货」按序处理，Kafka 只能保证分区内有序，所以顺序性完全落在分区键上。

按 orderId 做 key：同订单严格有序，不同订单可并行，这是订单事件的正确解。
按 skuId 做 key：看似能让库存事件聚在一起，但大促爆款会让单个 sku 的所有消息挤进同一分区，形成热点分区，其余分区闲置，lag 全堆在一个分区上——这就是本次事故里「扩容无效」的真正原因。
不设 key（轮询分区）：吞吐最高，但同订单乱序，退款可能先于支付被消费，库存直接被扣成负数。

还有一个必须提前想清楚的问题：不要在业务高峰期给 topic 增加分区。分区数变化会改变 key % partitions 的映射关系，正在进行中的订单事件会落到新分区上，此前积累的顺序保证当场失效。要扩容分区，就在低峰期做，并接受一个短暂的乱序窗口。

五、毒消息与死信：别让一条脏数据堵住整个分区

批量消费有个副作用：一批 200 条里只要有一条 JSON 解析失败，整批都会失败并持续重试，消费位移永远停在它前面，这就是典型的「毒消息堵分区」。位移提交语义上是「前面的都处理完了」，所以一条卡住等于整条链路停摆。

@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<String, String> template) {
    // 死信分区与原分区保持一致，方便定位
    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
            template,
            (record, ex) -> new TopicPartition(record.topic() + ".DLQ", record.partition()));

    // 重试 3 次，间隔 1s，仍然失败就投递死信
    DefaultErrorHandler handler =
            new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3));

    // 反序列化/参数错误属于不可恢复异常，不重试直接进死信
    handler.addNotRetryableExceptions(IllegalArgumentException.class);
    return handler;
}

把不可恢复异常加入 not-retryable 名单非常关键。默认情况下所有异常都会重试，遇到格式错误的消息就是三次无意义的重试加一轮日志刷屏，而消费线程被实实在在地占用着。

六、积压治理的正确顺序

先定位分区：用 kafka-consumer-groups.sh --describe --group stock-consumer 看每个分区的 LAG 与 CONSUMER-ID，判断是全局积压还是单分区热点。
再加并发：只在分区数有余量的情况下提高 concurrency，或者临时起一个同 group 的扩容实例。超过分区数没有任何收益。
降级非核心逻辑：积分、埋点这类旁路消费可以先关掉或异步化，把单批耗时压到原来的三分之一。
最后才考虑跳过：确认真实业务损失可接受时，用 seek() 跳到最新位移属于「丢数据」操作，必须留审批记录，绝不能当成常规手段。

监控上建议至少埋四个指标：消费组 lag 绝对值与增速、每秒提交位移数、rebalance 次数、单批处理耗时 P99。前两个看趋势，后两个是故障的先行指标——rebalance 次数出现异常抬升时，lag 通常还没开始涨，这时介入成本最低。

七、踩坑清单

批次事务边界不清：ack.acknowledge() 必须放在数据库事务提交之后。放在事务方法内部，事务还没提交就提交了位移，一旦回滚就是永久丢数据。
重试放大：消费端调下游 HTTP 接口时如果自带 3 次重试，再叠加 Kafka 的 3 次重试，一条消息会被打 9 次。要么统一在消费层重试，要么在下游关闭重试。
批量太大反而更慢：max-poll-records 从 200 调到 2000，单批耗时可能冲破 max.poll.interval.ms，重新掉进 rebalance 陷阱。批量大小要按「单批耗时的 P99 乘以 2 仍小于 poll interval」来定。
幂等表没有清理策略：consumed_event 按天分区或定期清理 7 天前数据，否则半年后这张表比业务表还大，索引维护成本会拖垮整批耗时。
反序列化异常被吞：EventCodec.decode 里如果 catch 住异常返回 null，坏消息会被静默丢弃，排查时完全看不到痕迹。应当抛出并进入死信。

八、小结

Kafka 消费积压的处置，本质上不是「调大并发」这么简单，而是先把三个语义厘清：位移什么时候提交、重复消费靠什么挡住、同一订单的顺序由谁保证。这三点定下来，吞吐优化才有意义。反过来说，如果先加并发再补幂等，只会把一次可恢复的延迟事故，放大成一次不可追溯的数据事故。