如何逐个消息发送给Kafka

Question

@Outgoing( "my-topic" )
public PublisherBuilder<Message<MessageWrapper>> generate() {
     if(Objects.isNull(currentMessage)){
          //currentMessage is an instance variable which is null when I start the application
          return ReactiveStreams.of(new MessageWrapper()).map(Message::of);
     }
     else {
          //currentMessage has been correctly set with the file information
          LOGGER.info(currentMessage);
          return ReactiveStreams.of(currentMessage).map(Message::of);
     }
}

如何保持一个连续的流，以“反应”新的丢弃的文件？（或其他事件，如HTTP GET请求或类似的事件）...

例如，如果我不返回PublisherBuilder的实例，而是返回一个整数，那么我的kafka主题将由一个非常巨大的整数值流填充。这就是为什么示例在发送消息时使用一些间隔...

我应该使用一些CompletationStage或CompletableFuture吗？RXJava2？使用哪个lib有点混乱(vertx，smallrye，rxjava2，microprofile，...)

null

非常感谢！

Answer 1

让我们从smallrye-reactive-messaging和smallrye-reactive-streams-operators之间的区别开始：smallrye-reactive-streams-operators与smallrye-reactive-messaging相同，但它还支持微轮廓上下文传播。由于大多数反应消息传递提供者在后台使用Vert.x，它将以事件循环样式处理您的消息，这意味着它将在单独的线程中运行。有时需要将一些ctx从基线程传播到新线程中（例如：填充CDI和Tx上下文以执行一些JPA实体管理器逻辑）。这里有ctx传播帮助。

用于方法签名。您可以查看SmallRye-reactive-streams的官方文档第3、4和5节。每一个都有不同的用例。你想用哪种口味的，由你决定。

什么时候用什么？如果您不是在反应上下文中运行，您可以使用下面的命令发送消息。

对于消息消费,您可以使用方法签名,如下所示：

@Incomputing（“Channel-2”）public CompletionStage doSomething（Message an

或