如何强制CompletableFuture.thenApply（）在运行前一阶段的相同线程上运行？

Question

以下是我面临的问题的简短代码版本：

public static void main(String[] args) {
    CompletableFuture.supplyAsync(() -> {
                /*
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException ignored) {}
                */
                //System.out.println("supplyAsync: " + Thread.currentThread().getName());
                return 1;
            })
            .thenApply(i -> {
                System.out.println("apply: " + Thread.currentThread().getName());
                return i + 1;
            })
            .thenAccept((i) -> {
                System.out.println("accept: " + Thread.currentThread().getName());
                System.out.println("result: " + i);
            }).join();
}

这是我得到的输出：

apply: main
accept: main
result: 2

我很惊讶地看到main在那里！当我取消注释Thread.sleep（）调用，甚至取消注释单个sysout语句时，我预计会发生这样的事情：

supplyAsync: ForkJoinPool.commonPool-worker-1
apply: ForkJoinPool.commonPool-worker-1
accept: ForkJoinPool.commonPool-worker-1
result: 2

我理解thenApplyAsync（）将确保它不会在主线程上运行，但是我想避免将供应商返回的数据从运行的线程传递到将要运行的线程然后在链中应用其他后续的代码。

Answer 1

方法thenApply计算调用方线程中的函数，因为未来已经完成。当然，当你在供应商中插入一个睡眠时，未来还没有完成，thenApply被调用。即使是print语句也可能会减慢供应商的速度，使主线程首先调用thenApply和thenAccess。但这不是可靠的行为，在重复运行代码时可能会得到不同的结果。

未来不仅不记得是哪个线程完成了它，而且也无法告诉任意线程执行特定代码。线程可能忙于其他事情，完全不合作，甚至在此期间终止。

只考虑

java prettyprint-override">ExecutorService s = Executors.newSingleThreadExecutor();
CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> {
    System.out.println("supplyAsync: " + Thread.currentThread().getName());
    return 1;
}, s);
s.shutdown();
s.awaitTermination(1, TimeUnit.DAYS);
cf.thenApply(i -> {
    System.out.println("apply: " + Thread.currentThread().getName());
    return i + 1;
})
.thenAccept((i) -> {
    System.out.println("accept: " + Thread.currentThread().getName());
    System.out.println("result: " + i);
}).join();

我们怎么能期望传递给然后apply和然后accept的函数在已经终止的池的工作线程中执行呢？

我们也可以写

CompletableFuture<Integer> cf = new CompletableFuture<>();

Thread t = new Thread(() -> {
    System.out.println("completing: " + Thread.currentThread().getName());
    cf.complete(1);
});
t.start();
t.join();

System.out.println("completer: " + t.getName() + " " + t.getState());
cf.thenApply(i -> {
    System.out.println("apply: " + Thread.currentThread().getName());
    return i + 1;
})
.thenAccept((i) -> {
    System.out.println("accept: " + Thread.currentThread().getName());
    System.out.println("result: " + i);
}).join();

它将打印相似的内容

completing: Thread-0
completer: Thread-0 TERMINATED
apply: main
accept: main
result: 2

显然，我们不能坚持让这个线程处理后续阶段。

但是，即使该线程是池中仍处于活动状态的工作线程，它也不知道它已经完成了未来，也不知道“处理后续阶段”的概念。在Executor抽象之后，它刚刚从队列中接收到一个任意的Runnable，在处理它之后，它继续其主循环，从队列中获取下一个Runnable。

因此，一旦第一个未来已经完成，告诉它完成其他未来工作的唯一方法就是让任务排队。当使用然后使用applyasync指定相同的池或在没有执行器的情况下使用…Async方法执行所有操作（即使用默认池）时，会发生这种情况。

当您对所有使用单线程执行器时...异步方法，您可以确保所有操作都由同一个线程执行，但它们仍然会通过池的队列。即使如此，在已经完成的未来情况下，主线程实际上也在排队执行相关操作，线程安全队列以及同步开销是不可避免的。

但是请注意，即使在单个工作线程顺序处理所有依赖操作之前，您首先创建了依赖操作链，这种开销仍然存在。每个未来的完成都是通过以线程安全的方式存储新状态来完成的，使结果对所有其他线程都可能可见，并自动检查是否同时发生了并发完成（例如取消）。然后，由其他线程链接的依赖操作将在执行之前以线程安全的方式获取。

所有这些具有同步语义的操作都使得在拥有一系列依赖的CompletableFutures时，由同一线程处理数据不太可能有好处。

要使实际的本地处理具有潜在的性能优势，唯一的方法是使用

CompletableFuture.runAsync(() -> {
    System.out.println("supplyAsync: " + Thread.currentThread().getName());
    int i = 1;

    System.out.println("apply: " + Thread.currentThread().getName());
    i = i + 1;

    System.out.println("accept: " + Thread.currentThread().getName());
    System.out.println("result: " + i);
}).join();

或者，换句话说，如果您不希望分离处理，那么首先不要创建分离处理阶段。