.switchIfEmpty() 得到热切评估有什么意义？答案

【问题标题】：What's the point of .switchIfEmpty() getting evaluated eagerly?.switchIfEmpty() 得到热切评估有什么意义？
【发布时间】：2020-01-12 04:56:48
【问题描述】：

即使我的流不为空，也总是会创建备用流？这样做的目的是什么？这是非常不习惯的。

另一方面，.onErrorResume 的评估是惰性的。

有人可以向我解释为什么 .switchIsEmpty 会被热切评估吗？

代码如下：

  public static void main(String[] args) {
    Mono<Integer> m = Mono.just(1);
    m.flatMap(a -> Mono.delay(Duration.ofMillis(5000)).flatMap(p -> Mono.empty()))
        .switchIfEmpty(getFallback())
        .doOnNext(a -> System.out.println(a))
        .block();
  }

  private static Mono<Integer> getFallback() {
    System.out.println("In Here");
    return Mono.just(5);
  }

输出是：

In Here (printed immediately)
5 (after 5s)

【问题讨论】：

您可能犯了在返回switchIfEmpty 使用的流之前进行初始化的错误。请提供您遇到问题的代码。
@akarnokd 我已经添加了代码。谢谢。
如果你只在main 中写getFallback(); 而没有switchIfEmpty 和其他结构会怎样？为什么？
方法在主线程上执行？看起来我缺少一些基本的东西。

标签： java reactive-programming project-reactor reactive

【解决方案1】：

如果你在它周围加上括号，为什么它会在其他任何地方执行？这种类型的误解经常出现，并且不确定来源是什么。

当你的代码被重写时，发生的事情应该会变得更加明显：

Mono<Integer> m = Mono.just(1);
Mono<Integer> m2 = m.flatMap(a -> Mono.delay(Duration.ofMillis(5000))
                                      .flatMap(p -> Mono.empty()));

Mono<Integer> theFallback = getFallback(); // <------------------ still on the main thread!

m2.switchIfEmpty(theFallback)
    .doOnNext(a -> System.out.println(a))
    .block();

getFallback 运行是因为它的父方法正在那里执行。这与反应式编程无关，而是大多数编程语言的基本属性。

【讨论】：

【解决方案2】：

这里需要了解的是组装时间和订阅时间的区别。

组装时间是您通过构建运营商链来创建管道的时间。此时您的发布者尚未订阅，您需要考虑一下。

订阅时间是您通过订阅触发执行并且数据开始流经您的管道的时间。这是你需要在回调、lambda、延迟执行等方面进行反应性思考的时候。

Simon Baslé 的 great article 中对此进行了详细介绍。

正如@akarnokd 在他的回答中提到的，getFallback() 方法在汇编时被强制调用，因为它没有被定义为 lambda，只是一个常规的方法调用。

您可以通过以下方法之一实现真正的懒惰：

1，您可以使用Mono.fromCallable 并将您的日志放入lambda：

public static void main(String[] args) {
    Mono<Integer> m = Mono.just(1);

    m.flatMap(a -> Mono.delay(Duration.ofMillis(5000)).flatMap(p -> Mono.empty()))
     .switchIfEmpty(getFallback())
     .doOnNext(a -> System.out.println(a))
     .block();
}

private static Mono<Integer> getFallback() {
    System.out.println("Assembly time, here we are just in the process of creating the mono but not triggering it. This is always called regardless of the emptiness of the parent Mono.");
    return Mono.fromCallable(() -> {
        System.out.println("Subscription time, this is the moment when the publisher got subscribed. It is got called only when the Mono was empty and fallback needed.");
        return 5;
    });
}

2，您可以使用Mono.defer 并将您的内部 Mono 的执行和组装延迟到订阅：

public static void main(String[] args) {
    Mono<Integer> m = Mono.just(1);
    m.flatMap(a -> Mono.delay(Duration.ofMillis(5000)).flatMap(p -> Mono.empty()))
     .switchIfEmpty(Mono.defer(() -> getFallback()))
     .doOnNext(a -> System.out.println(a))
     .block();
}

private static Mono<Integer> getFallback() {
    System.out.println("Since we are using Mono.defer in the above pipeline, this message gets logged at subscription time.");
    return Mono.just(5);
}

请注意，您的原始解决方案也非常好。您只需要知道返回 Mono 之前的代码是在汇编时执行的。

【讨论】：