Spring @KafkaListener消费百万消息性能低、被踢消费组问题咨询

两个问题本质是同一个原因导致的：

• Kafka消费者本身是单线程设计，默认情况下@KafkaListener的处理逻辑直接跑在消费者poll线程上，业务处理多久，poll循环就会被阻塞多久。默认配置下两次poll最大间隔是5分钟（max.poll.interval.ms=300000），超过这个时间broker就会判定消费者死亡，把它踢出消费组触发重平衡，这就是你看到报错的直接原因。
• 你单条消息处理要1-1.5秒，默认一次poll最多拉500条，极端情况这批消息全部处理完要12.5分钟，远超过5分钟的超时阈值，必然频繁触发重平衡。重平衡的时候整个消费组完全停摆，重平衡结束后之前没提交偏移量的消息会重新拉取消费，大量重复工作直接把吞吐量拖垮，这就是你现在每分钟只能处理90条的核心原因。
• 你现在concurrency配64是刚好的，和分区数匹配，不用改——Kafka规定一个分区同一时间只能给消费组里的一个消费者消费，concurrency开超过64的话，多出来的消费者永远分不到分区，纯浪费资源。

核心思路是poll线程和业务处理线程隔离，保证poll线程不被业务逻辑阻塞，同时通过可控的业务线程池提升处理吞吐量，具体步骤如下：

第一步：调整基础消费配置 #

先修改Spring Kafka的默认配置，给后续多线程处理留出容错空间，配置项如下：

spring:
  kafka:
    listener:
      # 开启批量消费，提升处理效率
      type: BATCH
      # 开启手动立即提交偏移量，避免丢消息
      ack-mode: MANUAL_IMMEDIATE
    consumer:
      # 单次poll最大拉取条数从默认500降到20，控制单批消息最大处理时长
      max-poll-records: 20
      # poll间隔超时调到30分钟，给业务处理留足够窗口，避免误踢
      properties:
        max.poll.interval.ms: 1800000

第二步：配置独立业务线程池 #

不要让业务逻辑跑在poll线程上，单独定义业务处理线程池。线程数按业务目标计算即可：目标吞吐量为100万条/小时≈278条/秒，单条处理1.2秒的话，总工作线程数需要278*1.2≈334个，预留冗余后配350核心线程就够。别一上来开上千个线程，你有数据库插入操作，线程数要和数据库连接池匹配，比如数据库连接池最大开400，业务线程池最多开350，剩下的连接留给其他组件用，不然全卡在拿连接上，开再多线程也没用。
线程池配置代码：

@Configuration
public class KafkaConsumerConfig {
    @Bean("bizProcessExecutor")
    public ThreadPoolExecutor bizProcessExecutor() {
        return new ThreadPoolExecutor(
                350,
                350,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                new ThreadFactoryBuilder().setNameFormat("kafka-biz-%d").build(),
                // 队列满时由poll线程自行处理任务，触发自然限流，避免OOM
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }
}

拒绝策略选CallerRunsPolicy不是随便选的：要是队列满了直接丢消息或者抛异常业务直接就炸了，让poll线程自己跑处理逻辑的话，poll线程被占住就不会继续拉新消息，相当于自动限流，不会把内存撑爆。

第三步：改造监听器逻辑 #

监听器中仅做消息分发，等整批消息全部处理完成后再提交偏移量，绝对不要提前提交，否则消费者宕机或重平衡时会直接丢失未处理的消息：

@Slf4j
@Service
public class BizKafkaListener {
    @Resource
    private ThreadPoolExecutor bizProcessExecutor;

    @KafkaListener(id = "xyz-listener", concurrency = "64", topics = "my-topic")
    public void onMessage(List<String> records, Acknowledgment ack) throws ExecutionException, InterruptedException {
        // 分发所有消息到业务线程池
        List<CompletableFuture<Void>> processFutures = records.stream()
                .map(record -> CompletableFuture.runAsync(() -> {
                    try {
                        // 替换为你的实际业务处理逻辑
                        doProcessRecord(record);
                    } catch (Exception e) {
                        log.error("消息处理失败, msg:{}", record, e);
                        // 此处自行实现重试、死信队列投递逻辑，不要让单条异常阻断整批
                    }
                }, bizProcessExecutor))
                .collect(Collectors.toList());
        // 等待整批消息全部处理完成
        CompletableFuture.allOf(processFutures.toArray(new CompletableFuture[0])).get();
        // 所有消息处理成功后再提交偏移量
        ack.acknowledge();
    }

    private void doProcessRecord(String record) {
        // 原有业务处理、数据库插入逻辑
    }
}

注意单条消息处理失败必须捕获异常，不要让单条失败导致整批消息永远无法提交偏移量，进入重复消费死循环。

第四步：配套性能优化 #

这部分能直接把处理吞吐量提升5-10倍，必须做：

• 把单条数据库插入改成批量插入：你单条处理1-1.5秒的耗时大概率来自单条事务提交的网络IO、磁盘IO开销，攒批后单次提交100-200条数据，平均单条处理耗时能降到100ms以内。
• 校验数据库连接池大小：确保连接池最大连接数大于业务线程池总线程数，避免线程因拿不到连接阻塞。
• 如果后续业务量上涨，优先优化单条消息处理逻辑，不要盲目增大线程数，避免数据库、下游服务被打垮。

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