我正在尝试使用库 reactor-kafka
开发一个 Spring Boot 应用程序,以对从 Kafka 主题读取的一些消息使用react。
我有一个构建KafkaReceiver
的配置类。
@Configuration
public class MyConfiguration {
@Bean
public KafkaReceiver<String, String> kafkaReceiver() {
Map<String, Object> props = new HashMap<>();
// Options initialisation...
final ReceiverOptions<String, String> receiverOptions =
ReceiverOptions.<String, string>create(props)
.subscription(Collections.singleton(consumer.getTopic()));
return KafkaReceiver.create(receiverOptions);
}
}
那么……现在呢?使用不太 react 的 spring-kafka
库,我可以用 @KafkaListener
注释一个方法,Spring Boot 会为我创建一个从 Kafka 主题监听的线程。
我应该把 KafkaReceiver
放在哪里?在所有示例中,我发现直接使用 main
方法,但这不是 Boot 方式。
我正在使用 Spring Boot 2.1.3 和 Reactor-Kafka 1.1.0
提前致谢。
最佳答案
既然你有那个 KafkaReceiver
bean,现在你可以这样做:
@Bean
public ApplicationRunner runner(KafkaReceiver<String, String> kafkaReceiver) {
return args -> {
kafkaReceiver.receive()
...
.sunbscribe();
};
}
ApplicationRunner
bean 将在 ApplicationContext
准备就绪时被踢出。有关详细信息,请参阅其 JavaDocs。
关于java - 整合Spring Boot和Reactor-Kafka的KafkaReceiver,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/55242567/