Kafka 消息积压如何处理？ - 面试题

Kafka 消息积压处理

Kafka 消息积压是生产环境中常见的问题，通常表现为 Consumer 消费速度跟不上 Producer 生产速度，导致消息在 Broker 中堆积。处理消息积压需要从多个维度进行分析和优化。

消息积压的原因

1. Consumer 消费能力不足

单线程消费：Consumer 使用单线程处理消息，处理速度慢
处理逻辑复杂：消息处理逻辑复杂，耗时较长
外部依赖慢：依赖外部系统（数据库、API 等）响应慢
资源限制：CPU、内存、网络等资源不足

2. Producer 生产速度过快

突发流量：短时间内大量消息涌入
生产配置不当：批量发送配置导致瞬时流量大
业务高峰期：业务高峰期消息量激增

3. 系统故障

Consumer 故障：Consumer 宕机或网络中断
依赖系统故障：数据库、缓存等依赖系统故障
网络问题：网络延迟或带宽不足

监控和诊断

监控指标

bash
# 查看 Consumer Lag
kafka-consumer-groups --bootstrap-server localhost:9092 \
  --describe --group my-group

# 查看消息积压情况
kafka-run-class kafka.tools.GetOffsetShell \
  --broker-list localhost:9092 \
  --topic my-topic --time -1

关键指标

Consumer Lag：Consumer 消费延迟
Messages Per Second：每秒消息数量
Bytes Per Second：每秒字节数
Log Size：日志文件大小

解决方案

1. 增加 Consumer 数量

原理：通过增加 Consumer 实例数量提高消费能力

实施步骤：

java
// 创建多个 Consumer 实例
for (int i = 0; i < consumerCount; i++) {
    ConsumerThread thread = new ConsumerThread(properties);
    thread.start();
}

注意事项：

Consumer 数量不能超过 Partition 数量
每个 Consumer 至少分配一个 Partition
合理设置 Consumer Group 的 Consumer 数量

2. 增加 Partition 数量

原理：通过增加 Partition 数量支持更多 Consumer 并行消费

实施步骤：

bash
# 增加 Topic 的 Partition 数量
kafka-topics --bootstrap-server localhost:9092 \
  --alter --topic my-topic --partitions 20

注意事项：

增加 Partition 不会重新分配现有消息
新消息会分配到新的 Partition
需要重启 Consumer 才能生效

3. 优化消费逻辑

批量处理：

java
// 批量处理消息
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
List<Record> batch = new ArrayList<>();
for (ConsumerRecord<String, String> record : records) {
    batch.add(record);
    if (batch.size() >= BATCH_SIZE) {
        processBatch(batch);
        batch.clear();
    }
}

异步处理：

java
// 使用线程池异步处理
ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
for (ConsumerRecord<String, String> record : records) {
    executor.submit(() -> processMessage(record));
}

优化外部依赖：

使用连接池
增加缓存
批量操作数据库
异步调用外部 API

4. 调整 Consumer 配置

properties
# 增加每次拉取的消息数
max.poll.records=1000

# 增加拉取间隔
max.poll.interval.ms=300000

# 增加会话超时时间
session.timeout.ms=30000

# 增加心跳间隔
heartbeat.interval.ms=3000

5. 临时扩容方案

创建临时 Topic：

bash
# 创建临时 Topic
kafka-topics --bootstrap-server localhost:9092 \
  --create --topic my-topic-temp --partitions 50 \
  --replication-factor 3

迁移消息：

bash
# 使用 MirrorMaker 迁移消息
kafka-run-class kafka.tools.MirrorMaker \
  --consumer.config consumer.properties \
  --producer.config producer.properties \
  --whitelist my-topic

增加临时 Consumer：

部署大量临时 Consumer
快速消费积压消息
消费完成后下线临时 Consumer

6. 丢弃非关键消息

选择性消费：

java
// 只消费最新的消息
consumer.seekToEnd(partitions);

// 跳过积压的消息
long currentOffset = consumer.position(partition);
long targetOffset = currentOffset - SKIP_COUNT;
consumer.seek(partition, targetOffset);

注意事项：

仅适用于非关键业务
需要评估数据丢失的影响
建议先备份积压的消息

预防措施

1. 容量规划

评估业务峰值流量
预留足够的 Consumer 实例
合理设置 Partition 数量

2. 监控告警

设置 Consumer Lag 告警阈值
监控消息积压趋势
及时发现和处理问题

3. 限流策略

java
// Producer 端限流
RateLimiter rateLimiter = RateLimiter.create(1000); // 1000 msg/s
rateLimiter.acquire();
producer.send(record);

4. 降级策略

在高峰期降级非核心功能
减少消息处理复杂度
使用简化逻辑处理消息

最佳实践

合理规划资源
- 根据 QPS 评估所需 Consumer 数量
- 预留 20-30% 的资源缓冲
- 考虑业务高峰期的流量
优化消费逻辑
- 简化消息处理逻辑
- 使用批量处理提高效率
- 减少外部依赖的调用
建立监控体系
- 实时监控 Consumer Lag
- 设置多级告警机制
- 定期检查系统健康状态
制定应急预案
- 准备临时扩容方案
- 建立消息备份机制
- 制定降级策略
定期演练
- 模拟消息积压场景
- 验证扩容方案
- 测试应急预案

通过综合运用以上方案，可以有效解决 Kafka 消息积压问题，确保系统的稳定性和可靠性。