Kafka消费者如何提升消息处理吞吐量？

面试场景

面试官：”你项目中的Kafka消费速度跟不上生产速度，造成消息堆积，你怎么优化？”

这道题考察对Kafka消费者模型的理解和调优能力。

消费者架构原理

┌─────────────────────────────────────────────────────────┐
│                    消费者组（Consumer Group）             │
├─────────────────────────────────────────────────────────┤
│                                                         │
│   Consumer-1       Consumer-2       Consumer-3         │
│       │                │                │               │
│   Partition-0     Partition-1     Partition-2          │
│                                                         │
└─────────────────────────────────────────────────────────┘

关键点：

一个分区只能被组内一个消费者消费
消费者数量 ≤ 分区数量才有意义

优化策略一：增加分区和消费者

扩展分区

# 增加分区数
kafka-topics.sh --alter --topic my-topic \
  --partitions 12 --bootstrap-server localhost:9092

增加消费者

// Spring Kafka配置多个消费者
@KafkaListener(topics = "my-topic", concurrency = "6")
public void consume(ConsumerRecord<String, String> record) {
    // 处理消息
}

注意：消费者数量不要超过分区数，否则多余的消费者会空闲。

优化策略二：批量消费

单条消费 vs 批量消费

// 单条消费（默认）
@KafkaListener(topics = "my-topic")
public void consume(ConsumerRecord<String, String> record) {
    process(record);  // 每条消息一次处理
}

// 批量消费
@KafkaListener(topics = "my-topic")
public void consumeBatch(List<ConsumerRecord<String, String>> records) {
    // 批量处理，减少网络往返
    batchProcess(records);
}

配置批量消费

spring:
  kafka:
    consumer:
      max-poll-records: 500  # 一次poll最多拉取500条
    listener:
      type: batch  # 开启批量消费模式

批量入库示例

@KafkaListener(topics = "order-topic")
public void consumeBatch(List<ConsumerRecord<String, String>> records) {
    List<Order> orders = records.stream()
        .map(r -> JSON.parseObject(r.value(), Order.class))
        .collect(Collectors.toList());
    
    // 批量插入，比单条插入快10倍以上
    orderMapper.batchInsert(orders);
}

优化策略三：多线程消费

问题

Kafka消费者poll是单线程的，处理逻辑慢会拖慢整体。

解决方案：业务处理异步化

@Service
public class OrderConsumer {
    
    private ExecutorService executor = Executors.newFixedThreadPool(20);
    
    @KafkaListener(topics = "order-topic")
    public void consume(ConsumerRecord<String, String> record) {
        // 提交到线程池异步处理
        executor.submit(() -> {
            try {
                processOrder(record);
            } catch (Exception e) {
                log.error("处理失败", e);
                // 失败处理：重试队列或死信
            }
        });
    }
}

注意事项

位移提交：异步处理需要手动提交位移
顺序性：多线程会打乱消息顺序
背压控制：线程池队列满时需要阻塞

完整的多线程消费方案

@Service
public class OrderConsumer {
    
    // 有界队列 + CallerRunsPolicy实现背压
    private ExecutorService executor = new ThreadPoolExecutor(
        20, 50, 60, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(1000),
        new ThreadPoolExecutor.CallerRunsPolicy()  // 队列满时由消费线程执行
    );
    
    @KafkaListener(topics = "order-topic")
    public void consume(List<ConsumerRecord<String, String>> records, 
                        Acknowledgment ack) {
        CountDownLatch latch = new CountDownLatch(records.size());
        
        for (ConsumerRecord<String, String> record : records) {
            executor.submit(() -> {
                try {
                    processOrder(record);
                } finally {
                    latch.countDown();
                }
            });
        }
        
        // 等待所有任务完成后提交位移
        latch.await();
        ack.acknowledge();
    }
}

优化策略四：参数调优

核心参数

参数	默认值	调优值	说明
fetch.min.bytes	1	1024	最小拉取字节数
fetch.max.wait.ms	500	100	最大等待时间
max.poll.records	500	1000	单次poll最大记录数
max.partition.fetch.bytes	1MB	2MB	单分区最大拉取字节

配置示例

spring:
  kafka:
    consumer:
      fetch-min-size: 1024        # 累积1KB再拉取
      fetch-max-wait: 100         # 最多等100ms
      max-poll-records: 1000      # 单次拉取1000条
      properties:
        max.partition.fetch.bytes: 2097152  # 2MB

优化策略五：避免Rebalance

Rebalance的影响

消费者Rebalance期间整个消费组暂停消费，严重影响吞吐。

常见触发原因

消费者心跳超时
消费处理时间过长
消费者加入/离开

参数优化

spring:
  kafka:
    consumer:
      properties:
        session.timeout.ms: 30000        # 会话超时
        heartbeat.interval.ms: 10000     # 心跳间隔
        max.poll.interval.ms: 600000     # 处理超时（10分钟）

优化效果对比

优化手段	吞吐量提升
增加分区+消费者	线性提升
批量消费	3-5倍
多线程处理	5-10倍
参数调优	20%-50%

面试答题框架

消费堆积原因分析：
- 消费者数量不足
- 处理逻辑太慢
- 网络/IO瓶颈

优化方案：
1. 增加分区+消费者（水平扩展）
2. 批量消费（减少网络往返）
3. 多线程处理（并行化）
4. 参数调优（fetch/poll参数）
5. 避免Rebalance（调大超时时间）

注意事项：
- 消费者数量≤分区数
- 多线程需要处理顺序和位移提交
- 批量处理需要考虑事务边界

总结

策略	核心思想	适用场景
增加分区	水平扩展	消费者数量已达上限
批量消费	减少IO次数	可接受少量延迟
多线程	并行处理	处理逻辑CPU密集
参数调优	减少等待	微调优化