能否动态配置Micronaut Kafka消费者的@Topic而非预定义？

当然可行，针对你的需求，有两种常用实现方式：

方式一：通过SpEL读取配置/Bean属性（启动时确定） #

如果topicName是固定从配置文件读取，或者通过构造函数注入的Bean属性，直接用Spring表达式语言(SpEL)在@Topic注解中引用即可：

场景1：读取配置文件属性 #

修改代码如下：

@KafkaListener(offsetReset = OffsetReset.EARLIEST)
public class KafkaConsumer {

    @Topic("${kafka.consumer.target-topic}")
    public void receive(@KafkaKey String day, String message) {
        System.out.println("Got Message for the  - " + day + " and Message is  " + message);
    }

}

然后在配置文件（如application.properties）中添加对应配置：

kafka.consumer.target-topic=你的目标topic名称

场景2：引用当前Bean的属性 #

如果一定要用构造函数传入的topicName字段，可直接通过SpEL引用当前Bean的属性：

@KafkaListener(offsetReset = OffsetReset.EARLIEST)
public class KafkaConsumer {

    private final String topicName;

    public KafkaConsumer(String topicName) {
        this.topicName = topicName;
    }

    @Topic("#{this.topicName}")
    public void receive(@KafkaKey String day, String message) {
        System.out.println("Got Message for the  - " + day + " and Message is  " + message);
    }

}

注意：这种方式要确保KafkaConsumer Bean初始化时，topicName已经被正确赋值。

方式二：动态注册KafkaListener（运行时动态切换） #

如果需要在程序运行过程中动态添加/切换topic消费者，就不能依赖静态的@Topic注解了，需要手动注册监听器：

@Component
public class DynamicKafkaConsumer {

    private final KafkaListenerEndpointRegistry registry;
    private final KafkaTemplate<String, String> kafkaTemplate;

    public DynamicKafkaConsumer(KafkaListenerEndpointRegistry registry, KafkaTemplate<String, String> kafkaTemplate) {
        this.registry = registry;
        this.kafkaTemplate = kafkaTemplate;
    }

    public void registerConsumer(String topicName) {
        // 创建监听器端点
        MethodKafkaListenerEndpoint<String, String> endpoint = new MethodKafkaListenerEndpoint<>();
        endpoint.setId("dynamic-consumer-" + topicName);
        endpoint.setTopics(topicName);
        endpoint.setOffsetResetStrategy(OffsetReset.EARLIEST);
        // 绑定消息处理方法
        try {
            endpoint.setMethod(this.getClass().getMethod("receive", String.class, String.class));
        } catch (NoSuchMethodException e) {
            throw new RuntimeException(e);
        }
        endpoint.setBean(this);
        // 注册并启动监听器
        registry.registerListenerContainer(endpoint, kafkaTemplate.getConsumerFactory());
    }

    public void receive(@KafkaKey String day, String message) {
        System.out.println("Got Message for the  - " + day + " and Message is  " + message);
    }
}

之后在业务逻辑中调用registerConsumer("目标topic名称")，就能动态创建对应topic的消费者。

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