使用@KafkaListener的topicPattern匹配时如何动态设置关联topic的消费组名称

首先明确一点：@KafkaListener的所有注解属性都必须是编译期常量，所以你没法直接在注解的groupId里写表达式动态获取匹配到的主题名，必须通过自定义监听注册逻辑实现，以下是最常用的可落地实现方案： #

自定义Kafka监听容器，按主题动态注册 #

你可以通过KafkaListenerEndpointRegistry手动为每个匹配到的tenant开头的主题单独注册监听容器，每个容器可以自定义独立的groupId，完全满足动态生成的需求：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.config.MethodKafkaListenerEndpoint;
import org.springframework.kafka.listener.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.stereotype.Component;
import jakarta.annotation.PostConstruct;
import java.util.List;

@Component
public class TenantKafkaListenerRegister {

    private final KafkaListenerEndpointRegistry endpointRegistry;
    private final ConcurrentKafkaListenerContainerFactory<String, String> containerFactory;

    // 构造注入所需依赖
    public TenantKafkaListenerRegister(KafkaListenerEndpointRegistry endpointRegistry,
                                       ConcurrentKafkaListenerContainerFactory<String, String> containerFactory) {
        this.endpointRegistry = endpointRegistry;
        this.containerFactory = containerFactory;
    }

    @PostConstruct
    public void registerAllTenantListeners() {
        // 1. 从Kafka集群拉取所有tenant开头的主题，可根据业务调整获取逻辑
        List<String> allTenantTopics = listTenantTopicsFromKafka();
        
        // 2. 为每个主题单独注册监听容器，动态生成groupId
        for (String topic : allTenantTopics) {
            MethodKafkaListenerEndpoint<String, String> endpoint = new MethodKafkaListenerEndpoint<>();
            endpoint.setId("tenant-listener-" + topic);
            endpoint.setTopics(topic);
            // 按规则动态生成消费组ID
            endpoint.setGroupId("my_group_" + topic);
            // 绑定消费逻辑
            endpoint.setMessageListener((MessageListener<String, String>) this::handleTenantMessage);
            // 启动容器
            endpointRegistry.registerListenerContainer(endpoint, containerFactory, true);
        }
    }

    // 从Kafka集群查询所有tenant开头的主题
    private List<String> listTenantTopicsFromKafka() {
        // 可通过AdminClient实现，示例逻辑如下：
        // AdminClient adminClient = AdminClient.create(containerFactory.getConsumerFactory().getConfigurationProperties());
        // return adminClient.listTopics().names().get()
        //         .stream()
        //         .filter(topicName -> topicName.startsWith("tenant"))
        //         .toList();
    }

    // 你的消费业务逻辑
    private void handleTenantMessage(ConsumerRecord<String, String> record) {
        // 处理消费消息的逻辑写在这里
    }
}

补充说明 #

• 该方案下每个租户主题对应独立的消费组，offset单独存储，天然满足多租户隔离的需求
• 如果需要支持后续新增tenant主题自动监听，只需要加定时任务或者监听Kafka主题新增事件，动态注册新的容器即可

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