Spring项目中如何配置多个Kafka消费者组

以下是两种常用实现方案，可根据业务场景选择：

方案1：同消费逻辑复用，多@KafkaListener注解叠加 #

如果多个消费者组的消费逻辑完全一致，直接在原有监听方法上新增@KafkaListener注解，每个注解对应不同的消费者组ID即可，代码改动量最小。

首先修改application配置（以yml为例），新增其他消费者组的配置项：

test:
  kafka:
    consumer:
      groupid: group-01 # 原有消费者组
      groupid-second: group-02 # 新增消费者组
  topics: test-topic1,test-topic2
  concurrency: 3

修改监听器代码，叠加注解即可：

// 原有消费者组监听
@KafkaListener(
        topics = {"#{'${test.topics}'.split(',')}"},
        groupId = "${test.kafka.consumer.groupid}",
        concurrency = "${test.concurrency}",
        containerFactory = "ListenerContainerFactory"
)
// 新增消费者组监听
@KafkaListener(
        topics = {"#{'${test.topics}'.split(',')}"},
        groupId = "${test.kafka.consumer.groupid-second}",
        concurrency = "${test.concurrency}",
        containerFactory = "ListenerContainerFactory"
)
void messageReceiver(@Payload List<String> data) {
    System.out.println("Message read start:");
}

方案2：不同消费逻辑/不同消费配置，配置多监听容器工厂 #

如果不同消费者组需要独立的消费配置（比如不同的并发数、反序列化规则、偏移量重置策略、重试规则），可以单独配置多个KafkaListenerContainerFactory，每个工厂对应一个消费者组的默认配置。

首先新增第二个监听容器工厂的配置：

@Configuration
public class KafkaConsumerConfig {
    // 原有容器工厂，对应group-01
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> ListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setBatchListener(true);
        // 其他原有配置
        return factory;
    }

    // 新增第二个容器工厂，对应group-02的独立配置
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> SecondListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        // 单独给第二个消费者组设置自定义配置
        Map<String, Object> configs = new HashMap<>(consumerFactory.getConfigurationProperties());
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "group-02");
        configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        DefaultKafkaConsumerFactory<String, String> secondConsumerFactory = new DefaultKafkaConsumerFactory<>(configs);
        factory.setConsumerFactory(secondConsumerFactory);
        factory.setConcurrency(5); // 独立并发数
        factory.setBatchListener(true);
        return factory;
    }
}

然后写两个独立的监听方法，分别绑定对应的容器工厂：

// 原有消费逻辑，对应group-01
@KafkaListener(
        topics = {"#{'${test.topics}'.split(',')}"},
        concurrency = "${test.concurrency}",
        containerFactory = "ListenerContainerFactory"
)
void firstGroupReceiver(@Payload List<String> data) {
    System.out.println("Group1 消费消息");
}

// 第二个消费逻辑，对应group-02
@KafkaListener(
        topics = {"#{'${test.topics}'.split(',')}"},
        containerFactory = "SecondListenerContainerFactory"
)
void secondGroupReceiver(@Payload List<String> data) {
    System.out.println("Group2 消费消息");
    // 可实现完全不同的消费逻辑
}

注意事项 #

• 不同消费者组的偏移量独立存储，每个组都会消费主题的全量消息，互相无影响
• 同消费逻辑场景优先选择方案1，代码更简洁
• 多容器工厂方案支持每个消费者组完全独立的配置和消费逻辑，灵活性更高

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