萃取器在反应器堵塞中的作用（从性能上看）

Question

我注意到不建议使用这样的提取器mono.toFuture()、flux.collectlist()，因为它们会阻塞流。

我不太确定“封锁”是以哪种方式进行的。就像下面的代码一样，我知道flux.collectlist()将等待所有项的完成，它是像某个线程一样等待，还是只是最后一个线程在最后完成.collectlist()的事情？

有人认为mono.toFuture()也会阻塞，它是否会立即返回一个“future”（当onnext()或oncomplete()发生时，该future将可用），或者直到onnext()或oncomplete()发生时才返回？

        var m = Flux.range(0, 100)
                .parallel()
                .runOn(Schedulers.boundedElastic())
                .map(i -> Mono.fromFuture(
                        Mono.just(i).map(n -> {
                            try {
                                var s = (long) (Math.random() * 100);
                                Thread.sleep(s);
                                System.out.println(Thread.currentThread() + "after " + s + "ms awaking: " + n);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            return n;
                        }).toFuture())
                )
                .doOnNext(o -> System.out.println(Thread.currentThread() + "before sequential"))
                .sequential();

        var mm = Flux.merge(m)
                .doOnNext(o -> System.out.println(Thread.currentThread() + "before collecting"))
                .collectList()
                .doOnNext(o -> System.out.println(Thread.currentThread() + "before map"))
                .map(list -> list.stream().map(i -> i).collect(Collectors.toList()))
                .publishOn(Schedulers.single())
                .toFuture();

Answer 1

你的假设不太正确。

mono.toFuture（）根本不是阻塞--它只是返回一个CompleteableFuture，您可以阻塞它（如果您调用它的get()方法）或者异步执行它（如果您使用它的任何异步方法，如thenApply()、thenCompose()等），您可以跳出Reactor上下文，从而丢失反压力之类的东西，但不必立即阻塞。

您可能会想到reactor的（非常）旧版本，我认为这里有一个toFuture（）变体返回Future，而不是CompleteableFuture-虽然这也没有阻塞，但它将您置于一个必须阻塞的上下文中，因为Future没有异步组件。因此，虽然方法调用本身没有阻塞，但这是您唯一的选择。

与普遍的看法相反，flux.collectlist()也不阻塞--它专门返回一个单字 ，这是一个将发出单个元素的非阻塞发布服务器，该元素是该通量中所有内容的列表。当然，您可以在此发布服务器上调用block()，该操作将是阻塞--但是单独调用collectlist()并不比任何其他运算符更阻塞。

话虽如此，它肯定会引起问题。由于它所做的工作的性质（将流量中的所有元素收集到内存中的单个列表中），它可能并不理想：

• 您可能需要等待很长时间才能发出列表，而没有关于它包含多少元素的反馈，也没有关于它是否被填充的反馈；
• 如果流量中的元素数或元素大小特别大，则可能会耗尽内存；
• 当添加元素时，您不能输出任何中间状态，因此您将失去流式JSON支持等功能。

然而，这并不使它阻塞，这只是意味着在决定是否值得在PariuClar场景中使用操作符之前，您需要权衡一组不同的潜在问题。