java - mono.zip 功能未按预期工作

Question

 Mono.zip(
            Mono.fromCallable(() -> queueDao.getNumberOfMessageInQueue(cartDaemonUrl)),
            Mono.fromCallable(() -> queueDao.getNumberOfMessageInQueue(orderConfirmationDaemonUrl))
            //ono.fromCallable(()->queueDao.getNumberOfMessageInQueue(cartDaemonUrl))
    )
            .map(T -> updateQueueCountToDb(T.getT1().block(), T.getT2().block())).
            doOnSuccess(row -> log.info("Queue info inserted into db rows ")).
            doOnError(e -> log.error("Error while inserting data stacktrace{}", e.getStackTrace()));

我无法弄清楚为什么控件没有进入 updateQueueCountToDb 方法。我也在该方法中添加了日志，即使这些日志没有打印在控制台中。

Answer 1

继续@Jesper 给出的评论，这是正确的，您必须订阅响应式(Reactive)流才能将它们付诸行动，否则，我们只是在构建我们期望数据流向的响应式(Reactive)管道。

我也在学习 react 流，因此，我在下面创建了一个快速示例，应该可以帮助您:

Mono<String> first=Mono.just("first");
Mono<String> second=Mono.just("second");

Mono<Tuple2<String, String>> zipped=Mono.zip(first,second);

zipped.map(tuple->someOtherOperation(tuple.getT1(),tuple.getT2()))
      .doOnSuccess(s->System.out.println("Success"))
      .doOnError(s->System.out.println("Error"))
      .subscribe();

我的 someOtherOperation 实现是:

public static Mono<Void> someOtherOperation(String a, String b) 
{
    System.out.println("Performing Operation "+a+":"+b);
    return Mono.empty();
}

因此，在 java 应用程序中运行我的代码片段将在控制台上打印以下内容:

Performing Operation first:second
Success

此外，subscribe 并不是订阅响应式(Reactive)流的唯一方法，请查看此文档 http://projectreactor.io/docs/core/release/reference/#_simple_ways_to_create_a_flux_or_mono_and_subscribe_to_it