Reactor单核与完全未来

Question

刚刚开始探索reactor项目及其抽象、Mono和Flux，并希望了解与Java8 barebones CompletableFuture的基本区别。

下面是我的一个简单代码：

public static void main(String[] args) throws Exception {

    Mono.fromCallable(() -> getData())
            .map(s -> s + " World ")
            .subscribe(s -> System.out.println(s));

    CompletableFuture.supplyAsync(() -> getData())
            .thenAccept(System.out::println);

    System.out.println(Thread.currentThread()+" End ");
}

private static String getData() {

    int j=0;

    for(int i=0; i<Integer.MAX_VALUE; i++){
        j = j - i%2;
    }

    System.out.println(Thread.currentThread()+" - "+j);
    return " Hello ";
}

首先，CompletableFuture并不奇怪。SupplyAsync通过ForkJoinPool调度函数的执行，“end”行立即打印，程序终止，因为主线程在这里确实很短--正如预期的那样。

但是mono.fromcallable(...)将主线程阻塞在那里。此外，在getData()函数中打印的线程名是主线程。因此，我看到的是顺序/阻塞行为，而不是顺序/非阻塞（异步）行为。是因为我在同一个线程上应用了订阅函数，所以它阻塞了吗？有人能解释一下吗？

Answer 1

是因为我在同一个线程上应用了订阅函数，所以它阻塞了吗？

这正是似乎发生的事情。

这种特定的行为让我有点惊讶，因为这不是大多数管道的行为方式。大多数管道都有这样或那样的方式，其中的一些操作使管道异步。publishon、subscribeon是明显的例子，但flatmap也可能具有这样的效果，可能还有许多其他效果。在这些情况下，subscribe将立即返回。

不过，这提示了反应式编程的一个非常重要的问题：管道不应该包含长的阻塞调用。准备一个反应管道，并在订阅时不阻塞地处理事件。因此，阻塞语句有可能阻塞整个执行。通过使用调度器，您可以将此类调用限制在特殊的线程池中，从而控制它们的效果。