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重新评估 lambda 的内容：

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 时，它不会立即创建它，它会在每个订阅中创建一个新的 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 立即创建了 observable 并且即使值已更改它也不会更改，但是 defer 在 subscribe 中创建 observable 所以您将使用当前的 onSubscribe 值

Answer 3

我正在尝试defer不同的用例。编写以下代码来检查和共享，因为它可能对其他人有所帮助。我的用例是链接两个Monos 并确保第一个在第二个被占用之前完成。第二个包含一个阻塞调用，其结果用于响应Monowithempty或errorresponse。没有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

Answer 4

初学者的简单答案：

当在 monoJust 变量上调用 subscribe 时，它将打印一个随机整数三次。但是在对 monoDefer 变量调用 subscribe 时，它每次都可以打印一个随机数。

   Mono<Integer> justMono = Mono.just((new Random()).nextInt(10));

    //this will print same random number thrice

    for(int i=0;i<3;i++)

        justMono.subscribe(x -> {System.out.println("Just Mono: " + x);});

    Mono<Integer> deferMono = Mono.defer(() -> Mono.just((new Random()).nextInt(10)));

    //this might print three different random numbers

    for(int i=0;i<3;i++)

        deferMono.subscribe(x -> {System.out.println("Defer Mono: " + x);});

在 Mono.just() 中，实例化仅在第一次订阅发生时发生一次。在 Mono.defer() 中，每次调用订阅时都会发生实例化。

如需更多参考，请查看： https ://www.youtube.com/watch?v=eupNfdKMFL4&t= 381s at 3:15 mins