如何強制CompletableFuture.thenApply()在運行前一階段的同一執行緒上運行？-有解無憂

這是我面臨的問題的簡短代碼版本：

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以及then鏈中的其他后續s。

uj5u.com熱心網友回復：

該方法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，您可以確保所有操作都由同一個執行緒執行，但它們仍將通過池的佇列。從那時起，在已經完成的未來的情況下，主執行緒實際上將依賴的動作排入佇列，執行緒安全佇列以及同步開銷是不可避免的。

但請注意，即使您設法首先創建依賴操作鏈，在單個作業執行緒按順序處理它們之前，這種開銷仍然存在。每個未來的完成都是通過以執行緒安全的方式存盤新狀態來完成的，使結果可能對所有其他執行緒可見，并同時原子地檢查并發完成（例如取消）是否發生。然后，由其他執行緒鏈接的依賴操作將在執行之前以執行緒安全的方式被獲取。

所有這些具有同步語意的操作使得在具有依賴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();

或者，換句話說，如果您不想要分離處理，首先不要創建分離處理階段。

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/404454.html

標籤：

上一篇：確保回呼在Mongo節點驅動程式中完成

下一篇：如何同時運行兩部分代碼