集成弹簧靴和反应器-卡夫卡的卡夫卡接收机

Question

我正在尝试开发一个Spring Boot应用程序，使用该库对从Kafka主题中读取的一些消息做出反应。reactor-kafka

我有一个构建.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);

    } 

}

井。。。现在？使用不那么反应式的库，我可以用Spring Boot注释一个方法，Spring Boot将为我创建一个从Kafka主题收听的线程。spring-kafka@KafkaListener

我应该将 放置 在哪里？在所有示例中，我发现直接使用方法，但这不是Boot方式。KafkaReceivermain

我正在使用Spring Boot 2.1.3和Reacter-Kafka 1.1.0

Answer 1

既然你有那颗豆子，现在你可以这样做：KafkaReceiver@Beanpublic ApplicationRunner runner(KafkaReceiver<String, String> kafkaReceiver) {&nbsp; &nbsp; &nbsp; &nbsp; return args -> {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; kafkaReceiver.receive()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ...&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .sunbscribe();&nbsp; &nbsp; &nbsp; &nbsp; };}当准备好时，这个豆子将被踢。有关详细信息，请参阅其 JavaDocs。ApplicationRunnerApplicationContext