java - Project Reactor - 长期运行的反应式 Kafka 消费者

Question

有没有人有一个长期运行的反应式 Kafka 消费者的例子？https://github.com/reactor/reactor-kafka.git中的 SampleConsumer 在订阅后退出，这是预期的。但是我需要不断地接收来自我们的主题的消息，并且在发生故障时我应该重新连接（特别是连接问题）。

我确实通过https://github.com/CollaborationInEncapsulation/s1p-reactor-netty-kafka-twitter.git作为帮助提出示例但不是很成功。感谢帮助

Answer 1

将 SampleConsumer 转换为 LongRunningSampleConsumer 是一个简单的解决方法。更改是添加一个 blockLast 以无限期地阻止。我知道阻止不是实现我的目标的最佳方式，我愿意接受替代解决方案。

当前代码在我下面的 git 中。

https://github.com/schengalath/reactor-kafka/blob/master/reactor-kafka-samples/src/main/java/reactor/kafka/samples/LongRunningSampleConsumer.java