基于Spring的Kafka同组多消费者监听带前缀多主题的技术问询

这需求太常见了，完全可以通过Spring Kafka的原生特性轻松实现，我给你梳理一套完整的实现方案，包括代码示例和关键注意点：

你的场景刚好契合Kafka消费者组的负载均衡机制：

• 所有目标主题都是单分区，每个主题的分区是独立的消费单元
• 同一个消费者组的多个实例，Kafka会自动把不同的分区（对应不同主题）分配给不同的实例，实现主题级别的负载分散
• 借助Spring Kafka的topicPattern属性，用正则表达式匹配所有带统一前缀的主题，还能自动监听后续新增的符合规则的主题

1. 配置文件（application.yml） #

先配置Kafka消费者的基础参数，确保多个实例的group-id完全一致：

spring:
  kafka:
    consumer:
      bootstrap-servers: your-kafka-broker-address:9092
      group-id: unified-test-group  # 所有实例必须用同一个组ID
      auto-offset-reset: earliest  # 首次启动时从最早偏移量开始消费
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      concurrency: 1  # 因为每个主题只有一个分区，这里设置为1即可，避免单个实例抢占多个分区

2. 消费者类实现 #

用@KafkaListener的topicPattern属性指定正则表达式，匹配所有前缀为test-的主题：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
public class PatternBasedKafkaConsumer {

    // 正则匹配所有以"test-"开头的主题
    @KafkaListener(topicPattern = "test-.*", groupId = "${spring.kafka.consumer.group-id}")
    public void consumeMessage(
            String message,
            @Header("kafka_receivedTopic") String receivedTopic  // 获取消息来源的主题
    ) {
        // 可以用实例标识（比如端口）区分不同的消费者实例
        String instanceIdentifier = System.getProperty("server.port", "local-instance");
        System.out.printf("[实例%s] 收到来自主题[%s]的消息：%s%n", instanceIdentifier, receivedTopic, message);

        // 如果需要按主题做差异化处理，这里可以分支逻辑
        if ("test-x".equals(receivedTopic)) {
            processTestXMessage(message);
        } else if ("test-y".equals(receivedTopic)) {
            processTestYMessage(message);
        } else {
            processOtherTestTopics(message);
        }
    }

    // 主题test-x的专属处理逻辑
    private void processTestXMessage(String message) {
        // 业务逻辑代码
    }

    // 主题test-y的专属处理逻辑
    private void processTestYMessage(String message) {
        // 业务逻辑代码
    }

    // 其他符合前缀的主题处理逻辑
    private void processOtherTestTopics(String message) {
        // 业务逻辑代码
    }
}

• 组ID一致性：所有消费者实例必须使用同一个group-id，否则无法实现负载均衡
• 自动发现新增主题：topicPattern会自动监听后续新增的符合正则规则的主题，不需要重启消费者服务
• 分区与并发数：因为每个主题只有1个分区，所以消费者的concurrency参数设置为1即可，设置更大的值不会有额外收益（单个分区只能被一个消费者实例消费）
• 主题创建：确保Kafka集群的auto.create.topics.enable配置为true（自动创建主题），或者提前手动创建所有需要的test-*主题
• 偏移量管理：如果需要自定义偏移量管理，可以结合AckMode或者手动提交偏移量，默认的自动提交已经能满足大部分场景

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