Mono.defer（）有什么作用？

Question

我在一些Spring webflux代码中遇到了Mono.defer（）

我在文档中查找了该方法，但不理解其中的解释：

“创建一个Mono提供程序，该提供程序将提供目标Mono以便为每个下游的订阅者进行订阅”

请给我一个解释和一个例子。我可能会参考一堆Reactor示例代码（它们的单元测试？）的地方。

谢谢

Answer 1

这有点过分简化，但是从概念上讲，Reactor的来源要么是懒惰的，要么是渴望的。诸如HTTP请求之类的更高级的请求将被延迟评估。另一方面，最简单的人喜欢Mono.just或Flux.fromIterable渴望。

那样的话，我的意思是调用Mono.just(System.currentTimeMillis())将立即调用该currentTimeMillis()方法并捕获结果。所述结果仅在被订阅后才发出Mono。多次订阅也不会更改该值：

Mono<Long> clock = Mono.just(System.currentTimeMillis());
//time == t0

Thread.sleep(10_000);
//time == t10
clock.block(); //we use block for demonstration purposes, returns t0

Thread.sleep(7_000);
//time == t17
clock.block(); //we re-subscribe to clock, still returns t0

该defer运营商有没有使这个源懒惰，重新评估拉姆达的含量每有一个新的用户时间：

Mono<Long> clock = Mono.defer(() -> Mono.just(System.currentTimeMillis()));
//time == t0

Thread.sleep(10_000);
//time == t10
clock.block(); //invoked currentTimeMillis() here and returns t10

Thread.sleep(7_000);
//time == t17
clock.block(); //invoke currentTimeMillis() once again here and returns t17

Answer 2

如果您在第一个视图中看到简单的单词，它就像Mono.just（），但不是。当您运行Mono.just（）时，它会立即创建一个Observable（Mono）并重用它，但是当您使用defer时，它不会立即创建它，因此会在每个subscription中创建新的Observable。

一个用例来了解差异

    int a = 5;
@Override
public void run(String... args) throws Exception {

    Mono<Integer> monoJust = Mono.just(a);
    Mono<Integer> monoDefer = Mono.defer(() -> Mono.just(a));

    monoJust.subscribe(integer1 -> System.out.println(integer1));
    monoDefer.subscribe(integer1 -> System.out.println(integer1));

    a = 7;
    monoJust.subscribe(integer1 -> System.out.println(integer1));
    monoDefer.subscribe(integer1 -> System.out.println(integer1));
}

打印：5,5,5,7

如果您看到mono.just立即创建了可观察对象，即使值已更改，它也不会更改，但是defer在subscription中创建了可观察对象，因此您将使用当前的onsubscribe值

Answer 3

我正在尝试defer不同的用例。编写以下代码进行检查并分享，因为它可能对其他人有帮助。我的用例是链接两个Monos 并确保第一个在第二个被占用之前完成。第二个包含一个阻塞调用，其结果用于响应 withMono或empty响应error。如果没有defer，我的阻塞调用就会执行，而不管第一个调用的结果如何。但是，在使用defer阻塞调用时，仅在第一个完成时执行Mono。代码如下：

public static void main(String[] args) {
    long cur = System.currentTimeMillis();
    boolean succeed = true;

    Mono<Integer> monoJust = Mono.create(consumer -> {
        System.out.println("MonoJust inside " + (System.currentTimeMillis() - cur));
        if (succeed) {
            consumer.success(1);
        } else {
            consumer.error(new RuntimeException("aaa"));
        }
    });

    Mono<String> monoJustStr = Mono.create(consumer -> {
        System.out.println("MonoJustStr inside: " + (System.currentTimeMillis() - cur));
        consumer.success("one");
    });

    System.out.println("##1##: Begin");
    monoJust.then(evaluator() ? Mono.empty() : monoJustStr).subscribe(d -> System.out.println("##1##: "+d), e-> System.err.println(e));
    System.out.println("##1##: Done: "+(System.currentTimeMillis() - cur));

    System.out.println("\n\n\n##2##: Begin");
    monoJust.then(Mono.defer(() -> evaluator() ? Mono.empty() : monoJustStr)).subscribe(d -> System.out.println("##2##: "+d), e-> System.err.println(e));
    System.out.println("##2##: Done: " + (System.currentTimeMillis() - cur));

}

private static boolean evaluator() {
    System.out.println("Inside Evaluator");
    return false;
}

输出succeed=true- 观察“Inside Evaluator”和“MonoJust inside”的顺序

##1##: Begin
Inside Evaluator
MonoJust inside 540
MonoJustStr inside: 542
##1##: one
##1##: Done: 542



##2##: Begin
MonoJust inside 544
Inside Evaluator
MonoJustStr inside: 544
##2##: one
##2##: Done: 544

以下是输出succeed = false- 请注意，未调用评估器。

##1##: Begin
Inside Evaluator
MonoJust inside 565
java.lang.RuntimeException: aaa
##1##: Done: 567



##2##: Begin
MonoJust inside 569
java.lang.RuntimeException: aaa
##2##: Done: 569