Kafka消息积压（挤压）通常指消费者处理速度跟不上生产者发送速度，导致消息在Broker端堆积。其核心场景与解决方案如下：
一、消息积压的主要场景
| 消费者处理能力不足 | 1. 消费逻辑复杂：单条消息处理耗时过长（如数据库操作、复杂计算）。 2. 消费者数量不足：消费者实例数少于Topic分区数，导致部分分区无消费者。 3. 消费者性能瓶颈：机器资源（CPU、内存、IO）不足或JVM GC频繁。 | 这是最常见的原因，消费速度持续低于生产速度。 |
| 生产者流量激增 | 1. 业务高峰：如大促、秒杀活动，瞬时写入量远超日常。 2. 数据灌入：批量数据导入或补偿任务集中触发。 | 生产端流量突然暴涨，消费端来不及应对。 |
| 消费端故障 | 1. 消费者Bug：消费逻辑抛出异常，导致进程崩溃或持续失败。 2. 依赖服务异常：如数据库、缓存、下游接口不可用，消费线程阻塞。 3. 错误的重试策略：无限重试某条失败消息，阻塞后续消息。 | 消费端完全或部分停止工作，Lag（滞后）持续增长。 |
| Topic/分区规划不合理 | 1. 分区数过少：限制了消费的并行度上限。 2. 数据倾斜：少数分区承载了绝大部分流量，导致对应消费者压力过大。 | 系统架构设计存在瓶颈，无法水平扩展。 |
二、消息积压的解决方案
解决思路遵循“先止血，后根治”原则：先快速恢复业务，再优化系统防止复发。
1. 紧急扩容，快速消费（止血）
这是处理线上积压最直接有效的方法。
- 水平扩容消费者：紧急增加消费者实例数，确保消费者数 ≤ 分区数。对于Kafka，增加Consumer实例后，Rebalance机制会自动分配分区，提升整体消费能力。
- 临时提升消费者性能：优化消费者组机器的资源配置，或迁移到更高配的服务器。
- 将积压数据转移至新Topic：如果积压量巨大，可以写一个临时程序，将老Topic的积压数据快速消费并转发到一个具有更多分区的新Topic，然后启动大量消费者并行处理新Topic的数据。
2. 优化消费端逻辑（根治）
- 异步与批量处理：将耗时的同步操作改为异步，或采用批量处理提升吞吐。 // 示例：将单条入库改为批量入库
@KafkaListener(topics = "my-topic")
public void consume(List<ConsumerRecord<String, String>> records) {
List<MyEntity> entityList = new ArrayList<>();
for (ConsumerRecord<String, String> record : records) {
// 转换消息为业务对象 entityList.add(convertToEntity(record.value()));
}
// 批量写入数据库，极大提升效率 myRepository.saveAll(entityList);
} - 优化消费逻辑：检查并优化消费代码中的慢SQL、远程调用、循环等瓶颈。
- 完善监控与告警：监控Consumer Lag（滞后数）和消费TPS，设置阈值告警，以便在积压初期及时干预。
3. 调整Kafka与应用配置
- 增加分区数：如果消费瓶颈在于并行度，可以增加Topic的分区数，然后相应增加消费者实例。注意：增加分区数只对新数据生效，需要结合“转移至新Topic”方案处理已有积压。
- 优化消费者参数：
- fetch.min.bytes / fetch.max.wait.ms：适当调大，增加每次拉取的数据量，减少网络往返。
- max.poll.records：调大单次拉取的消息数，提高批处理效率。
- session.timeout.ms / max.poll.interval.ms：根据处理耗时合理设置，避免不必要的Rebalance。
4. 架构层面预防
- 容量规划与压测：根据业务峰值预估流量，提前对Kafka集群和消费服务进行压测，规划合理的分区数和资源。
- 设置死信队列（DLQ）：对于反复处理失败的消息，可将其投递到DLQ，避免阻塞主流程，事后由单独任务处理。
- 流量削峰：在生产者端或消费端前引入缓冲层（如Redis、内存队列），平滑突发流量。
- 降级与熔断：当依赖的下游服务异常时，消费端应有熔断机制，避免线程池被拖垮，并可执行降级策略（如将消息持久化后稍后重试）。
参考来源
- kafka面试之如何防止消息丢失、重复消费、消息挤压
- 【kafka】消息队列
- 【消息中间件】详解三大MQ：RabbitMQ、RocketMQ、Kafka
- Kafka-消息积压、消息过期、读写分离、一致性保障
- 消息中间件MQ——RabbitMQ、RocketMQ、Kafka
