我可以在运行时向我的@kafkalistener 添加主题吗答案

【问题标题】：Can i add topics to my @kafkalistener at runtime我可以在运行时向我的@kafkalistener 添加主题吗
【发布时间】：2021-10-15 01:52:40
【问题描述】：

我为主题数组创建了一个 bean，在运行时我向这个主题数组添加了一些主题，但消费者没有更新主题，并且仍然从主题数组中的第一个主题消费。我希望消费者添加这些新主题并开始从中消费

@Autowired
private String[] topicArray;

@KafkaListener(topics = "#{topicArray}", groupId = "MyGroup")
    public void listen(...) {
        ...
    }

【问题讨论】：

标签： spring-kafka

【解决方案1】：

没有；该属性在初始化期间被评估一次。

您不能在运行时向现有侦听器容器添加主题。

但是，您可以将侦听器 bean 设为原型 bean，并在每次您想要侦听新主题时创建一个新容器。

这是一个例子：

@SpringBootApplication
public class So68744775Application {

    public static void main(String[] args) {
        SpringApplication.run(So68744775Application.class, args);
    }

    private String[] topics;

    private final AtomicInteger count = new AtomicInteger();

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    Foo foo() {
        return new Foo();
    }

    @Bean
    Supplier<String> idProvider() {
        return () -> "so68744775-" + count.getAndIncrement();
    }

    @Bean
    Supplier<String[]> topicProvider() {
        return () -> this.topics;
    }

    @Bean
    ApplicationRunner runner(ApplicationContext context) {
        return args -> {
            this.topics = new String[] { "topic1", "topic2" };
            context.getBean(Foo.class);
            this.topics = new String[] { "topic3" };
            context.getBean(Foo.class);
        };
    }

}

class Foo {

    @KafkaListener(id = "#{idProvider.get()}", topics = "#{topicProvider.get()}", groupId = "grp")
    public void listen(String in) {
        System.out.println(in);
    }

}

不过，最好省略groupId，这样每个容器都在自己的组中（id 属性）。这避免了在添加新容器时不必要的重新平衡。

【讨论】：

它解决了我的问题，感谢@GaryRussell 的解释。