如何手动启动 Kafka 监听器？

Question

我有一些用@KafkaListener 注释的方法，但我只想手动启动其中一些（取决于某些条件）。

@KafkaListener(id = "consumer1", topics = "topic-name", clientIdPrefix = "client-prefix", autoStartup = "false")
public void consumer1(String message) {
    // consume
}

@PostConstruct
private void startConsumers() {
    if (true) {
        kafkaListenerEndpointRegistry.getListenerContainer("consumer1").start();
    }
}

但此时kafkaListenerEndpointRegistry.getListenerContainers() 是空列表，kafkaListenerEndpointRegistry.getListenerContainer("consumer1") 返回null。所以也许@PostConstruct 方法被调用的那一刻还为时过早，听众还没有注册。 我尝试用@Scheduled(fixedDelay = 100) 注释startConsumers() 方法，并且监听器已经可用。但是对于我想在启动应用程序后调用一次的东西，使用@Scheduled 并不是一个好的决定。

Answer 1

您不能在 @PostConstruct 中执行此操作 - 在应用程序上下文生命周期中还为时过早。

实现SmartLifecyle，将阶段设置为Integer.MAX_VALUE，并在start()方法中启动容器。

或者使用@EventListener 并监听ApplicationStartedEvent（如果使用Spring Boot）或ContextRefreshedEvent 用于非Boot Spring 应用程序。