如何解决CompletableFuture不同阶段在不同ExecutorService执行异常？

需求 #

我希望创建一个在指定ExecutorService上运行的CompletableFuture，并为其添加一个在另一个ExecutorService上执行的后续阶段。

已尝试代码 #

try (ExecutorService benchRunner = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() - 1);
     ExecutorService resultHandler = Executors.newVirtualThreadPerTaskExecutor()) {
    for (Algorithm algorithm : Algorithm.values()) {
        for (Distribution distribution : Distribution.values()) {
            for (Integer arraySize : List.of(30, 60, 90, 300, 600, 900, 3000, 6000, 9000)) {
                SortingBenchmark.Config config = new SortingBenchmark.Config(arraySize, algorithm, distribution);

                CompletableFuture<Benchmark> benchmarkTask = new CompletableFuture<>();
                benchmarkTask.completeAsync(new SortingBenchmark(config), benchRunner);
                benchmarkTask.whenCompleteAsync((b, t) -> {
                    if (t != null) {
                        System.err.println("Failed on: " + config);
                        return;
                    }

                    System.out.println("Finished: " + config);
                }, resultHandler);
            }
        }
    }
}

问题现象 #

我理解whenCompleteAsync应该在benchRunner上完成new SortingBenchmark(config)执行后，在resultHandler上运行。但实际运行时，whenCompleteAsync中的代码从未执行，而completeAsync部分运行正常（可通过7个核心占用100%及标准输出判断）。
若为benchmarkTask添加join调用，whenCompleteAsync的代码会开始运行，但此操作会导致整个流程变为单线程。

疑问 #

我是否误解了whenCompleteAsync的工作机制？该如何实现我的需求？

你的问题出在错误创建并使用CompletableFuture实例：你手动新建了CompletableFuture对象后调用completeAsync，但completeAsync只是异步完成这个手动创建的未完成Future，并非返回关联任务的新Future。更关键的是，try-with-resources块结束时会立即关闭两个ExecutorService，此时resultHandler还没机会执行后续任务，且你没有等待所有异步任务完成，直接导致后续阶段被中断或无法启动。

正确实现方式 #

不要手动创建CompletableFuture，直接用supplyAsync（带返回值任务）或runAsync（无返回值任务）创建初始任务，链式调用whenCompleteAsync，最后必须等待所有任务完成后再让ExecutorService关闭。

修改后的代码如下：

try (ExecutorService benchRunner = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() - 1);
     ExecutorService resultHandler = Executors.newVirtualThreadPerTaskExecutor()) {
    
    // 收集所有异步任务，用于后续统一等待
    List<CompletableFuture<Void>> allTasks = new ArrayList<>();

    for (Algorithm algorithm : Algorithm.values()) {
        for (Distribution distribution : Distribution.values()) {
            for (Integer arraySize : List.of(30, 60, 90, 300, 600, 900, 3000, 6000, 9000)) {
                SortingBenchmark.Config config = new SortingBenchmark.Config(arraySize, algorithm, distribution);

                // 用supplyAsync创建初始任务，指定benchRunner执行（假设SortingBenchmark实现Callable）
                CompletableFuture<Benchmark> benchmarkTask = CompletableFuture.supplyAsync(
                    () -> new SortingBenchmark(config).call(),
                    benchRunner
                );

                // 链式调用whenCompleteAsync，指定resultHandler执行后续逻辑
                CompletableFuture<Void> handlerTask = benchmarkTask.whenCompleteAsync((b, t) -> {
                    if (t != null) {
                        System.err.println("Failed on: " + config);
                        return;
                    }
                    System.out.println("Finished: " + config);
                }, resultHandler);

                allTasks.add(handlerTask);
            }
        }
    }

    // 等待所有任务完成，包括基准测试执行和后续处理
    CompletableFuture.allOf(allTasks.toArray(new CompletableFuture[0])).join();
}

关键说明 #

1. 使用标准API创建CompletableFuture：supplyAsync/runAsync是创建异步任务Future的标准方式，会自动关联任务与Future，避免手动创建带来的关联错误。
2. 必须等待所有任务完成：try-with-resources的自动关闭机制会在代码块结束时终止ExecutorService，若不等待任务完成，后续处理逻辑会被直接中断。CompletableFuture.allOf(...).join()可以等待所有异步任务（包括后续阶段）执行完毕。
3. 链式调用保证执行顺序：whenCompleteAsync会在前置任务完成后，自动在指定的resultHandler线程池中执行，完全符合你“初始任务在benchRunner执行，后续处理在resultHandler执行”的需求。

如果SortingBenchmark是Runnable而非Callable，只需将supplyAsync替换为runAsync，并调整后续逻辑即可。

内容的提问来源于stack exchange，提问作者ZarakshR