如何使用CompletableFuture.thenComposeAsync()？

Question

鉴于:

public class Test
{
    public static void main(String[] args)
    {
        int nThreads = 1;
        Executor e = Executors.newFixedThreadPool(nThreads);

        CompletableFuture.runAsync(() ->
        {
            System.out.println("Task 1. Thread: " + Thread.currentThread().getId());
        }, e).thenComposeAsync((Void unused) ->
        {
            return CompletableFuture.runAsync(() ->
            {
                System.out.println("Task 2. Thread: " + Thread.currentThread().getId());
            }, e);
        }, e).join();
        System.out.println("finished");
    }
}

我期待一个执行程序线程运行任务1,然后执行任务2.相反,代码挂起,如果nThreads小于2.

1. 请解释代码挂起的原因.我可以看到它在CompletableFuture中被阻止:616等待一些Future完成,但目前尚不清楚原因.
2. 如果我允许使用2个线程,那么每个线程用于什么？

总之,请帮助我了解thenComposeAsync()实际工作原理.Javadoc看起来像是为机器人而不是人类而写的:)

Answer 1

1. 该thenComposeAsync方法为您的执行程序设置了一个新任务,它抓取单个线程并等待您Task 2完成.但是这个没有更多的线程可以运行.您可以改为使用thenCompose在同一线程中执行的方法Task 1来避免死锁.

2. 一个线程正在执行Task 1,Task 2第二个线程正在处理组合两者的结果.

注意:CompletableFuture(s)ForkJoinPool在处理产生新任务的任务时更有效.默认ForkJoinPool用Java 8中添加了对这一目的,默认情况下使用,如果你不指定执行运行任务.

以下是关于这些新功能在何处发挥作用以及它们如何工作的良好演示:Java 8期货的反应式编程模式.

Answer 2

它阻塞在runAsync里面thenComposeAsync。 thenComposeAsync在执行程序e内的线程中运行提供的函数。但是您赋予它的函数会尝试在同一执行程序中执行runAsync的主体。

通过添加另一个跟踪输出，您可以更好地了解发生了什么：

CompletableFuture.runAsync(() -> {
    System.out.println("Task 1. Thread: " + Thread.currentThread().getId());
}, e).thenComposeAsync((Void unused) -> {
    System.out.println("Task 1 1/2. Thread: " + Thread.currentThread().getId());
    return CompletableFuture.runAsync(() -> {
        System.out.println("Task 2. Thread: " + Thread.currentThread().getId());
    }, e);
}, e).join();

现在，如果使用2线程执行程序运行它，您将看到Task 1 1/2和Task 2在不同的线程上运行。

解决它的方法是thenComposeAsync用常规替换thenCompose。我不确定你为什么会用thenComposeAsync。如果您有一个返回的方法，则该方法CompletableFuture可能不会阻塞并且不需要异步运行。