java 库，支持异步编写 futures (flatMap)

Question

我认为新的 java 8 CompletableFuture 将能够使用“thenComposeAsync”方法在不阻塞线程的情况下一个接一个地执行异步任务。

我认为事实并非如此。线程同步等待任务完成 (CompletableFuture.java:616 r = fr.waitingGet(false);)

如果 nThreads <= 2，下面的代码永远不会执行“任务 3”:

int nThreads= 2;
Executor e= Executors.newFixedThreadPool(nThreads);

CompletableFuture.runAsync( () -> {
    System.out.println("task 1 threadId " + Thread.currentThread().getId());
}, e).thenComposeAsync( (Void v1) -> {
    return CompletableFuture.runAsync( () -> {
        System.out.println("task 2 threadId " + Thread.currentThread().getId());
    }, e).thenComposeAsync((Void v2) -> {
        return CompletableFuture.runAsync( () -> {
            System.out.println("task 3 threadId " + Thread.currentThread().getId());
        }, e);
    }, e);
}, e).join();
System.out.println("finished");

我应该使用其他什么java库？我希望该代码在一个线程中执行(nThreads = 1)。

Answer 1

如果您使用 ForkJoinPool 作为 ExecutorService，则您的示例可以与 thenComposeAsync 配合使用:

final ExecutorService e = new ForkJoinPool(nThreads);

另外，最后一定要关闭服务(否则程序不会退出):

e.shutdown();

编辑:也许还有更多解释:如果您绘制包含由 thenComposeAsync 导致的任务的依赖关系图，这些任务必须在其后继者可以运行之前完成(因为它们提供传递给其后继者的返回值)，您会发现您需要两个以上的线程与对手调度程序一起执行图。

通常，很难预测执行动态创建的任务图最少需要多少个线程。 ForkJoinPool 通过尝试保持在给定的并行级别内来确认这一点，但如果需要的话会创建额外的线程。