Spring Boot下同一Group Id的多Kafka消费者并行消费实现问询

你不需要重复编写多个相同逻辑的@KafkaListener方法，Spring Kafka通过并发配置就能轻松实现同一消费组内的多消费者并行消费，以下是具体方案和配置细节：

一、核心配置：设置消费并发数 #

Spring Kafka的ConcurrentKafkaListenerContainerFactory支持通过concurrency属性指定并发消费者数量，容器会自动创建对应数量的消费者实例，每个实例独立处理分配到的分区。

方式1：在容器工厂Bean中全局配置 #

修改你的kafkaListenerContainerFactoryBean，添加setConcurrency()方法：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setCommitLogLevel(LogIfLevelEnabled.Level.DEBUG);
    factory.getContainerProperties().setMissingTopicsFatal(false);
    // 设置并发数，比如10个消费者实例
    factory.setConcurrency(10);
    return factory;
}

方式2：在@KafkaListener注解中局部配置 #

如果只想给特定Listener设置并发数，可以直接在注解上指定concurrency属性，优先级高于工厂的全局配置：

@RetryableTopic(
        attempts = "3",
        backoff = @Backoff(delay = 1000, multiplier = 2.0),
        autoCreateTopics = "false")
@KafkaListener(topics = "myTopic", groupId = "myGroupId", concurrency = "10")
public void consume(@Payload(required = false) String message) {
     processMessage(message);
}

二、关键注意事项 #

• 分区数与并发数的匹配：Kafka的并行消费能力上限由主题的分区数决定，建议并发数≤分区数。如果并发数大于分区数，多余的消费者实例会处于空闲状态；如果分区数大于并发数，单个消费者会分配到多个分区处理。
• @RetryableTopic的兼容：你使用了@RetryableTopic，需要确保重试主题（包括死信主题）的分区数与原主题myTopic一致，否则重试消息的并行处理会受限。
• 消费者配置的独立性：每个并发消费者实例都会使用consumerFactory创建，默认的DefaultKafkaConsumerFactory是线程安全的，无需额外处理。
• 消息顺序性：如果需要保证同一Key的消息顺序，要确保Kafka生产者按Key分区发送，同一Key的消息会被分配到同一个分区，由单个消费者顺序处理，不会被并发打乱。

三、其他优化建议 #

• 可以通过配置文件动态设置并发数，避免硬编码：

  spring:
    kafka:
      listener:
        concurrency: 10
  

  或者在工厂Bean中读取配置：

  factory.setConcurrency(Integer.parseInt(env.getProperty("kafka.listener.concurrency", "5")));
  

• 调整消费者的max.poll.records配置，控制每次拉取的消息数量，平衡吞吐量和处理效率：

  config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
  

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