CompletableFuture.all使用哪个执行程序？

Question

假设我们有两个执行人，1和2。

我们可以配置在执行时使用哪个执行器

CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(()-> {return 1;}, executor1) //executor1
CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(()-> {return 2;}, executor1) //executor1
CompletableFuture<Integer> cf3 = CompletableFuture.supplyAsync(()-> {return 3;}, executor2) //executor2

但哪个线程执行器使用CompletableFuture静态方法allOf？

CompletableFuture.allOf(cf1, cf2, cf3)

谢啦！

Answer 1

没有与CompletableFuture#allOf相关联的执行器，它只会生成CompletableFuture，它将在同一个线程中等待依赖项的完成，您将在该线程中调用CompletableFuture#get（）。

在您的示例中，cf1和cf2后面的任务仍将由执行器1执行，cf2中的任务将由执行器2执行，这是allOf（…）的结果。get（）将在当前线程中返回，并且不会在后台启动其他线程。

下面是一个示例，您可以通过在System.out.println行上设置断点并检查活动线程列表来观察IDE中的实际行为。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.function.Supplier;

import static java.util.concurrent.CompletableFuture.supplyAsync;

public class ExecutorTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Executor executor1 = Executors.newSingleThreadExecutor();
        Executor executor2 = Executors.newSingleThreadExecutor();
        CompletableFuture<Integer> cf1 = supplyAsync(run(1), executor1); //executor1
        CompletableFuture<Integer> cf2 = supplyAsync(run(2), executor1); //executor1
        CompletableFuture<Integer> cf3 = supplyAsync(run(3), executor2); //executor2
        CompletableFuture<Void> result = CompletableFuture.allOf(cf1, cf2, cf3);
        new Thread(() -> {
            try {
                result.get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }).start();
        System.out.println("Waiting now...");
    }

    private static Supplier<Integer> run(int result) {
        return () -> runDelayed(result);
    }

    private static int runDelayed(int result) {
        try {
            Thread.sleep(30000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return result;
    }

}

Answer 2

伊凡·加梅尔的答案并不准确。

实际上，没有与allOf（）返回的CompletableFuture相关联的执行器，因为事实上，从来没有与任何CompletableFuture相关联的执行器。

任务与执行程序相关联，因为它在其中运行，但关联是反向的：执行程序有一个要执行的任务列表。

任务还可以与可完成的未来相关联，该未来将在任务完成时完成。CompletableFuture本身不保留对用于创建它的任务或执行器的引用。然而，它可以保留对任务的引用，并可以选择在依赖阶段中使用的执行器。

由allOf（）返回的CompletableFuture将由任务完成，该任务是原始CompletableFuture的从属阶段。在您的示例中，此任务可以通过以下方式执行：

• 如果第三个任务首先完成，则执行者1
• 如果前两个任务在第三个任务之前完成，则执行器2；或
• 如果在调用allOf（）之前所有任务都已完成，则为原始线程

这可以通过向allOf（）调用添加依赖的thenRun（）阶段来看出：

public class CompletableFutureAllOfCompletion {
    private ExecutorService executor1 = Executors.newFixedThreadPool(2);
    private ExecutorService executor2 = Executors.newFixedThreadPool(2);
    private Random random = new Random();

    public static void main(String[] args) {
        new CompletableFutureAllOfCompletion().run();
    }

    public void run() {
        CompletableFuture<Integer> cf1 = supplyAsync(this::randomSleepAndReturn, executor1);
        CompletableFuture<Integer> cf2 = supplyAsync(this::randomSleepAndReturn, executor1);
        CompletableFuture<Integer> cf3 = supplyAsync(this::randomSleepAndReturn, executor2);
        randomSleepAndReturn();
        CompletableFuture.allOf(cf1, cf2, cf3)
                .thenRun(() -> System.out.println("allOf() commpleted on "
                        + Thread.currentThread().getName()));

        executor1.shutdown();
        executor2.shutdown();
    }

    public int randomSleepAndReturn() {
        try {
            final long millis = random.nextInt(1000);
            System.out.println(
                    Thread.currentThread().getName() + " waiting for " + millis);
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 0;
    }
}

pool-1-thread-1 waiting for 937
pool-1-thread-2 waiting for 631
main waiting for 776
pool-2-thread-1 waiting for 615
allOf() commpleted on pool-1-thread-1

pool-1-thread-1 waiting for 308
pool-1-thread-2 waiting for 788
main waiting for 389
pool-2-thread-1 waiting for 863
allOf() commpleted on pool-2-thread-1

pool-1-thread-1 waiting for 168
pool-1-thread-2 waiting for 292
main waiting for 941
pool-2-thread-1 waiting for 188
allOf() commpleted on main

由于无法保证将使用的执行器，因此对其中一个方法的调用之后应该是异步（，executor）调用，以控制将使用哪个执行器。

如果需要返回其中一个调用的结果CompletableFuture，只需添加一个applyasync（i-