Spring Boot用@KafkaListener消费每秒20万条消息是否需做并发优化？

结论 #

你当前默认配置的@KafkaListener完全无法承载每秒20万的消息流量，必须做并发、批量消费等相关优化。

核心问题说明 #

Spring Kafka默认的@KafkaListener是单线程消费模式，单个消费线程的处理能力通常在每秒数千到数万条之间，取决于你的过滤、转发逻辑耗时，和20万的QPS要求差距极大。

必须做的优化配置 #

• 首先确保你监听的Kafka Topic总分区数足够：Kafka的并发消费上限等于Topic的总分区数，假设单线程每秒能处理1万条消息，你至少需要预留20个以上的分区。
• 配置监听并发数：给@KafkaListener增加concurrency参数，数值不要超过监听Topic的总分区数，避免资源浪费，示例配置：

@KafkaListener(topics = "#{'${app.kafka.consumer.topic}'.split(',')}", concurrency = "20")
public void receivedMessage(ConsumerRecord<String, String> cr, @Payload String message){
    log.info("Message received from topic {} ", cr.topic());
    //TODO
}

• 开启批量消费：将单条消费模式改成批量消费，能大幅提升吞吐量，需要修改两处配置：
  1. application配置新增：spring.kafka.listener.type=batch，同时可调整spring.kafka.consumer.max-poll-records设置每次批量拉取的消息数量，建议根据业务耗时调整到500~2000区间
  2. 消费方法入参修改为批量类型：

@KafkaListener(topics = "#{'${app.kafka.consumer.topic}'.split(',')}", concurrency = "20")
public void receivedMessage(List<ConsumerRecord<String, String>> records){
    log.info("Batch received {} messages", records.size());
    // 批量过滤、批量转发
}

额外优化建议 #

• 过滤逻辑尽量轻量化，避免在消费线程中执行阻塞IO操作，如果有重耗时逻辑可以丢到独立线程池异步处理，注意做好offset提交控制，避免消息丢失
• 转发消息到其他Topic时使用KafkaTemplate的异步发送能力，不要同步等待发送结果，除非你有极强的消息可靠性要求
• 按需调整Consumer拉取配置：调整fetch.min.bytes、fetch.max.wait.ms参数，平衡消费延迟和吞吐量

内容的提问来源于stack exchange，提问作者ppb