在Spring Boot应用程序中实现反应式Kafka侦听器

Question

我试图在我的Spring Boot应用程序中实现反应性kafka消费者，我正在看这些例子：https://github.com/reactor/reactor-kafka/blob/master/reactor-kafka-samples/src/main/java/reactor/kafka/samples/SampleScenarios.java

看起来在被动Kafka中还没有对Spring的支持

我理解SpringKafka监听器如何在非反应式KafkaAPI中工作：最简单的解决方案是为ConcurrentKafkaListenerContainerFactory和ConsumerFactory配置bean，然后使用@KafkaListener annotation，瞧

但我不知道如何正确使用SpringKafka。

基本上我需要一个听众来讨论这个话题。我应该创建自己的循环或调度程序吗？或者我错过了什么。任何人都可以分享他们的知识和最佳实践吗？

Answer 1

我还没有现成的解决方案，但我正在尝试这个（Kotlin代码，Spring Boot）。有人在这里发布了部分代码片段https://github.com/reactor/reactor-kafka/issues/100

@EventListener(ApplicationStartedEvent::class)
fun onSomeEvent() {
    kafkaReceiver
        .receive()
        .doOnNext { record ->
            val myEvent = record.value()
            processMyEvent(myEvent).thenEmpty {
                record.receiverOffset().acknowledge()
            }
        }
        .doOnError {
            /* todo */
        }
        .subscribe()
}

查看其他堆栈溢出问题。虽然不多，但可能会给你一些想法

• 使用onErrorResume来处理问题负载发布到Kafka使用ReactorKafka
• 反序列化异常后继续消耗kafkaReactor中的后续记录