3

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

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()会确保它不会在main线程上运行，但我想避免将供应商返回的数据从运行supplyAsync的线程传递给将要运行的线程thenApply以及链中的其他后续thens。

4

1 回答 1

6

该方法thenApply评估调用者线程中的函数，因为未来已经完成。当然，当你插入一个sleep供应商，未来还没有完成的时候，thenApply被调用。即使是打印语句也可能会减慢供应商的速度，以使主线程thenApply首先调用thenAccept。但这不是可靠的行为，重复运行代码时可能会得到不同的结果。

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

只考虑

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();

我们怎么能期望传递给已经终止的池的工作线程thenApply并在其中执行的函数？thenAccept

我们也可以写

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从队列中获取下一个。

因此，一旦第一个 future 完成，告诉它完成其他 future 的工作的唯一方法就是将任务排入队列。当使用thenApplyAsync指定相同的池或使用…Async没有执行程序的方法执行所有操作时会发生这种情况，即使用默认池。

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

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

所有这些具有同步语义的操作使得在具有依赖链时由同一个线程处理数据不太可能有好处CompletableFuture。

进行可能具有性能优势的实际本地处理的唯一方法是使用

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();

或者，换句话说，如果您不想要分离处理，请不要首先创建分离处理阶段。

于 2022-01-05T16:48:01.873 回答