Project Reactor：等待 Mono 的列表完成

Question

我希望代码等待 Mono 完成然后收集结果，但它永远不会发生。

为什么？

这是我的代码：

public static void main(String[] args) throws Exception {
        Mono<Integer> mono1 = Mono.fromCallable(() -> 1);
        Mono<Integer> mono2 = Mono.fromCallable(() -> 2);
        List<Mono<Integer>> monos = Arrays.asList(mono1, mono2);

        Mono
                .when(monos)
                .subscribe(__ -> {
                    int i1 = mono1.block();
                    int i2 = mono1.block();

                    System.out.println(i1 + i2);
                });

        Thread.currentThread().join();
    }

Answer 1

您的 Mono.when() 调用是 Mono<Void> 类型的 - 它只是完成（或错误，如果它的发布者之一返回错误。）因此永远不会发出元素，因此永远不会调用 subscribe()，因此该订阅块中的代码永远不会执行。

目前尚不清楚您想在这里发生什么，但最快的“修复”可能是在订阅之前到materialize()，因此您会得到onComplete() 信号作为元素传播：

Mono
        .when(monos)
        .materialize()
        .subscribe(__ -> {
            int i1 = mono1.block();
            int i2 = mono1.block();

            System.out.println(i1 + i2);
        });

请注意，这将打印2，而不是3，因为i1 和i2 都引用mono1。