Reactor Mono vs CompletableFuture

Question

刚刚开始探索反应堆项目及其抽象Mono和Flux,并希望了解与java 8准系统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 将立即返回。

不过，这暗示了关于反应式编程的一个非常重要的点：管道不应包含长阻塞调用。反应式管道旨在准备好并在订阅时无阻塞地处理事件。因此，阻塞语句具有阻塞整个执行的真正潜力。使用 aScheduler您可以将此类调用限制在特殊的 ThreadPools 中，从而控制它们的效果。