外观
百万消息积压治理：从扩容到跳过的避坑实录
线上消息队列最危险的时刻，往往不是队列空转，而是监控图上的 Lag 从几万迅速爬升到百万。此时业务方催单、运维催恢复、群里不断有人建议“多起几个消费者”。但在 Kafka 等分区消息模型中，消费者数量并不等于消费能力：超过分区数的实例拿不到数据，频繁启停还可能触发再均衡，反而让所有消费者短暂停摆。
本文复盘一次百万级消息积压治理过程，重点说明何时扩容、如何安全再均衡、什么情况下允许跳过，以及跳过之后怎样补偿。示例以 Java + Spring Kafka 为主，配置和代码均可直接改造后运行。
问题背景：百万 Lag 是如何形成的
消息积压通常不是单一原因，而是“生产速率、消费速率、单条处理耗时、失败重试”共同作用的结果。典型链路如下：
那次故障的表象是订单状态通知延迟，根因有三个：第一，营销活动导致消息生产量短时翻倍；第二，消费者依赖的评价服务 RT 从 50ms 升到 2s；第三，失败消息在内存中无限重试，消费线程被毒丸消息占满。此时如果只看 CPU，会误判为应用负载不高；真正要看的指标是分区 Lag、单条处理耗时分位数、重试次数和下游依赖成功率。
方案对比：扩容、加分区、跳过分别适合什么场景
治理积压不能只有“重启”和“加机器”两种手段，至少要比较三类方案。
| 方案 | 适用场景 | 优点 | 风险 |
|---|---|---|---|
| 增加消费者实例 | 分区数充足、消费逻辑无状态、下游仍可承受 | 见效快，改动小 | 实例数超过分区数无效；频繁启停触发再均衡 |
| 增加 Topic 分区 | 长期流量增长，现有消费者并行度不足 | 提升长期吞吐上限 | 历史消息仍在旧分区；键顺序可能变化；需要扩容消费者 |
| 临时跳过坏消息 | 少量格式错误或不可恢复异常阻塞链路 | 快速恢复主链路 | 丢业务语义，必须有死信和补偿 |
判断顺序应当是：先确认瓶颈是否在消费者并行度，再决定扩容；如果分区数不足，先评估加分区；只有明确消息不可处理或允许延迟补偿时，才使用跳过策略。把跳过当成“静默吞异常”，一定会制造数据不一致。
消费者扩容：并行度不能超过分区数
Kafka 一个分区在同一消费组内只能被一个消费者实例消费。假设 Topic 有 12 个分区，把消费者从 3 个扩到 6 个通常有效，扩到 20 个则有 8 个实例空转。更重要的是，扩容前必须评估下游容量：消息处理本质是把压力传给数据库、缓存或 RPC 服务，消费者快了但下游被打垮，Lag 只是换一种方式继续涨。
下面是一份偏稳态的消费者配置：
yaml
spring:
kafka:
bootstrap-servers: ${KAFKA_BOOTSTRAP:127.0.0.1:9092}
consumer:
group-id: order-event-consumer
enable-auto-commit: false
# 积压时不要一次拉太多，避免一批消息处理超时被误判下线
max-poll-records: 100
session-timeout: 30s
heartbeat-interval: 10s
max-poll-interval: 300s
auto-offset-reset: latest
listener:
ack-mode: manual_immediate
type: batch配置含义是关闭自动提交，由业务处理成功后再确认位点；批量拉取但不要无限批量；心跳与会话超时分离，避免业务处理稍慢就被 Group 协调者踢出。
分区再均衡：避免扩容变成二次事故
默认消费组成员身份是动态的，实例启动、停止、网络抖动都可能触发再均衡。再均衡期间，分区会被撤销，消费者不能正常拉取消息。积压时如果一次性拉起几十个 Pod，或者没有优雅停机，就可能出现“扩容—再均衡—处理中断—再次心跳超时”的风暴。
实践中建议使用静态成员，并在监听器中处理分区撤销与提交：
java
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
public class SafeRebalanceListener implements ConsumerRebalanceListener {
private final ThreadLocal<Set<TopicPartition>> pendingPartitions =
ThreadLocal.withInitial(ConcurrentHashMap::newKeySet);
private final OffsetCommitter offsetCommitter;
public SafeRebalanceListener(OffsetCommitter offsetCommitter) {
this.offsetCommitter = offsetCommitter;
}
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
// 再均衡前尽量提交已处理完成的位点，避免撤销后重复消费扩大
offsetCommitter.commitProcessedOffsets(partitions);
pendingPartitions.get().removeAll(partitions);
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
pendingPartitions.get().addAll(partitions);
}
}部署侧还应配置滚动发布，而不是同时重建全部实例。Kubernetes 中可以让新实例分批加入，并保证应用收到 SIGTERM 后先停止拉取、等待正在处理的消息结束，再离开消费组。静态成员可通过如下参数开启：
yaml
spring:
kafka:
consumer:
properties:
group.instance.id: ${HOSTNAME:consumer}-${random.uuid}
session.timeout.ms: 30000
heartbeat.interval.ms: 10000
max.poll.interval.ms: 300000
partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignorCooperativeStickyAssignor 使用协作式再均衡，可以减少“全量撤销再分配”造成的停顿。需要注意，消费组内所有消费者应使用兼容的分配策略，否则可能出现协议不一致或回退到全量再均衡。
关键实现：限流、批处理、死信与跳过开关
积压治理的核心不是让消费者无限快跑，而是在下游可承受范围内稳定输出。下面示例展示批量消费、按异常分类跳过、写入死信表以及手动确认位点。
java
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
public class OrderEventConsumer {
private final OrderEventHandler handler;
private final DeadLetterService deadLetterService;
private final ChaosSwitch chaosSwitch;
public OrderEventConsumer(OrderEventHandler handler,
DeadLetterService deadLetterService,
ChaosSwitch chaosSwitch) {
this.handler = handler;
this.deadLetterService = deadLetterService;
this.chaosSwitch = chaosSwitch;
}
@KafkaListener(topics = "order-event", groupId = "order-event-consumer")
public void consume(List<ConsumerRecord<String, String>> records,
Acknowledgment ack) {
for (ConsumerRecord<String, String> record : records) {
try {
OrderEvent event = OrderEventParser.parse(record.value());
handler.handleWithRateLimit(event);
} catch (UnprocessableMessageException ex) {
// 明确不可恢复的消息：记录死信，不能静默丢弃
deadLetterService.save(record, "UNPROCESSABLE", ex.getMessage());
} catch (Exception ex) {
if (chaosSwitch.isSkipTemporaryEnabled()) {
// 紧急开关只用于故障窗口，必须带原因、批次和过期时间
deadLetterService.save(record, "TEMP_SKIP", ex.getMessage());
} else {
// 可恢复异常不提交位点，触发重试，避免污染后续位点
throw ex;
}
}
}
// 整批业务处理与死信落库成功后，才提交本批最后一个位点
ack.acknowledge();
}
}死信不要只发到另一个 Topic 后就无人管理，至少要持久化原始消息、Topic、分区、位点、异常类型、失败时间和重试状态。补偿任务可以按时间扫描死信表，在下游恢复后重新投递。
java
public void replayDeadLetters() {
List<DeadLetter> letters = deadLetterService
.findDueMessages("TEMP_SKIP", 100);
for (DeadLetter letter : letters) {
try {
OrderEvent event = OrderEventParser.parse(letter.getPayload());
handler.handleWithRateLimit(event);
deadLetterService.markSuccess(letter.getId());
} catch (Exception ex) {
// 控制重试次数，超过阈值进入人工处理，避免无限补偿
deadLetterService.markRetryFailed(letter.getId(), ex.getMessage());
}
}
}如果需要快速追赶历史消息，还可以临时启动独立消费组做“旁路消化”：主消费组继续处理实时消息，旁路组按业务优先级把旧消息写入补偿队列。但这要求处理逻辑具备幂等性，否则重复执行会造成订单状态反复变更。
避坑总结：百万积压中最常见的三个错误
第一，消费者实例数盲目超过分区数。现象是新 Pod 启动成功、日志显示加入消费组，但没有任何消息进入，监控吞吐也不增长。修正方法是先查看 Topic 分区数和消费组分配情况，确保实例数不大于分区数；长期并行度不足时再规划分区扩容，同时评估顺序键和历史消息迁移。
第二，处理一批消息时间过长，频繁触发再均衡。典型异常是消费者刚拉到几百条消息，处理数分钟后出现 Member ... sending LeaveGroup 或 max.poll.interval.ms 超时。修正方法是降低 max.poll.records，把耗时任务异步化或限流，给 max.poll.interval.ms 留出真实 P99 处理时间，并通过优雅停机保证撤销分区前提交位点。
第三，捕获异常后直接 continue，却把位点提交到最新位置。现象是 Lag 很快清零，但业务数据缺失，且因为位点已经前进，无法再从原 Topic 自动重放。修正方法是“跳过必落死信、落库成功再提交位点”，并为临时跳过开关设置审批、过期时间和补偿看板。
第四，只扩容消费者，不保护下游。现象是消息积压下降几分钟后，数据库连接池耗尽或评价服务线程池打满，最终消费再次停滞。修正方法是在消费者内设置并发上限、令牌桶限流和熔断降级，按下游容量而不是本机 CPU 决定消费速率。
消息积压治理的目标不是让监控曲线瞬间变零，而是在不破坏数据一致性的前提下恢复稳定处理。扩容解决并行度，再均衡策略保证扩缩容平稳，跳过策略只隔离不可处理或允许延迟的消息，最终仍要靠死信补偿闭环。把这四件事做扎实，百万 Lag 才不会从一次性能问题演变成长期数据事故。
免责声明
本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。