本文將深入講解 ExecutorCompletionService 的使用以及源碼解析。

ExecutorCompletionService適用場景

ExecutorCompletionService在以下場景中特別有用：

• 并行任務處理：當需要同時執(zhí)行多個任務，并按照完成的順序獲取它們的結(jié)果時，可以使用ExecutorCompletionService來簡化任務提交和結(jié)果獲取的流程。
• 高性能計算：在需要進行大規(guī)模計算或復雜計算的場景中，可以將任務拆分成多個子任務，并使用ExecutorCompletionService來管理和獲取子任務的結(jié)果。

假設現(xiàn)在有一批需要進行計算的任務，為了提高整批任務的執(zhí)行效率，我們可以使用線程池來異步計算這些任務。通過向線程池中不斷提交任務并保留與每個任務關(guān)聯(lián)的Future對象。最后，我們可以遍歷這些Future對象，并通過調(diào)用 get() 方法獲取每個任務的計算結(jié)果。

Future的不足

Future 沒有辦法回調(diào)，只能手動去調(diào)用，當通過 get() 方法獲取線程的返回值時，會導致阻塞，也就是和當前這個 Future 關(guān)聯(lián)的計算任務執(zhí)行完成的時候才返回結(jié)果，新任務必須等待已完成任務的結(jié)果才能繼續(xù)進行處理。

這樣會浪費很多時間，因為我們不知道哪個線程先執(zhí)行完了，只能挨個去獲取結(jié)果，這樣已經(jīng)完成的線程會因為前面未完成的線程的耗時而無法提前進行匯總，最好是誰先執(zhí)行完成，誰先返回。

而 ExecutorCompletionService 可以實現(xiàn)這樣的效果，節(jié)省獲取完成結(jié)果的時間，它的內(nèi)部有一個先進先出的阻塞隊列，用于保存已經(jīng)執(zhí)行完成的 Future，通過調(diào)用它的 take() 方法或 poll() 方法可以獲取到一個已經(jīng)執(zhí)行完成的 Future，進而通過調(diào)用 Future 接口實現(xiàn)類的 get() 方法獲取最終的結(jié)果。

CompletionService的目標是任務誰先完成誰先獲取，即結(jié)果按照完成先后順序排序

ExecutorCompletionService使用

ExecutorCompletionService 提供了一種方便的方式來處理一組異步任務，并按照完成的順序獲取它們的結(jié)果。它內(nèi)部使用了Executor框架來執(zhí)行任務，并且內(nèi)部管理著一個已完成任務的阻塞隊列，在結(jié)果獲取上提供了更加靈活和高效的機制。

下面是一個簡單的例子來演示ExecutorCompletionService的基本使用：

public class ExecutorCompletionServiceExample {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(5);
        CompletionService<String> completionService = new ExecutorCompletionService<>(executor);

        // 提交任務
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            completionService.submit(() -> {
                double sleepTime = Math.random() * 1000;
                Thread.sleep((long) sleepTime); // 模擬耗時操作
                return "Task " + taskId + " completed,cost time: " + sleepTime;
            });
        }

        // 獲取結(jié)果
        for (int i = 0; i < 10; i++) {
            Future<String> future = completionService.take();
            String result = future.get();
            System.out.println(result);
        }

        executor.shutdown();
    }
}

輸出：

Task 2 completed,cost time: 170.01927312611775
Task 3 completed,cost time: 460.9622858036789
Task 1 completed,cost time: 563.24738180643
Task 0 completed,cost time: 595.938819219159
Task 5 completed,cost time: 480.4473056068137
Task 4 completed,cost time: 748.2343208613524
Task 6 completed,cost time: 370.4679098376097
Task 7 completed,cost time: 270.45945981324905
Task 9 completed,cost time: 336.5536570760892
Task 8 completed,cost time: 577.5774464801026

在上述代碼中，我們創(chuàng)建了一個固定大小的線程池，并使用 ExecutorCompletionService 來提交和獲取任務的結(jié)果。通過調(diào)用completionService.submit()方法來提交任務，并隨機指定睡眠時間，來模擬任務執(zhí)行的耗時，然后通過completionService.take()方法來獲取已完成的任務結(jié)果。

可以看到是按照任務的執(zhí)行耗時順序去獲取結(jié)果的。

ExecutorCompletionService原理解析

ExecutorCompletionService 提供了兩個構(gòu)造函數(shù)，一個可以指定阻塞隊列，另一個使用內(nèi)部默認的阻塞隊列，兩個構(gòu)造函數(shù)都需要傳進線程池參數(shù)。

圖片

提供了三個獲取方法，可以看到都是從隊列中獲取。

• take()/poll() 方法的工作都委托給內(nèi)部的已完成任務隊列 completionQueue。
• 如果隊列中有已完成的任務, take() 方法就返回任務的結(jié)果，否則阻塞等待任務完成。
• poll() 與 take() 方法不同，poll() 有兩個版本:

無參的 poll() 方法：如果完成隊列中有數(shù)據(jù)就返回，否則返回null。

有參數(shù)的 poll() 方法：如果完成隊列中有數(shù)據(jù)就直接返回，否則等待指定的時間，到時間后如果還是沒有數(shù)據(jù)就返回null。

圖片

兩個提交任務方法，可以看到 submit() 方法最終會委托給內(nèi)部的 executor 去執(zhí)行任務，提交任務的時候會將任務封裝成 QueueingFuture 對象。

圖片

ExecutorCompletionService內(nèi)部維護了 QueueingFuture 類，QueueingFuture 繼承了 FutureTask，并重寫了 done() 方法，

可以看到 done() 方法在任務完成的時候會將結(jié)果存進 已完成任務隊列 completionQueue 中。

圖片

Futuretask 的 done() 方法是用來標記一個任務已經(jīng)完成的方法。當一個 Futuretask 中的任務完成后，就會調(diào)用 done() 方法通知。

圖片

默認是空方法，不會執(zhí)行任何動作。

圖片

執(zhí)行流程

當我們使用ExecutorCompletionService類時，它能夠按照任務完成的順序獲取它們的結(jié)果，這是因為ExecutorCompletionService類內(nèi)部結(jié)合了QueueingFuture類和done()方法的機制。以下是源碼流程步驟解釋：

1. 提交任務：

我們通過submit方法將任務提交給ExecutorCompletionService。在提交任務時，ExecutorCompletionService會使用自定義的QueueingFuture類來包裝任務，并將其交給底層線程池執(zhí)行。

2. QueueingFuture類：

QueueingFuture類是ExecutorCompletionService的內(nèi)部類，繼承自FutureTask。它的構(gòu)造方法接收一個Callable對象作為參數(shù)。

在QueueingFuture類中，它重寫了done()方法。done()方法會在任務執(zhí)行完成后被調(diào)用。

3. 任務執(zhí)行完成時的處理：

當任務執(zhí)行完成后，在底層線程池的Worker線程中，會調(diào)用QueueingFuture的done()方法。

在done()方法中，QueueingFuture會首先調(diào)用父類FutureTask的done()方法，以觸發(fā)對計算結(jié)果的獲取。然后，它會將任務的結(jié)果存儲到一個內(nèi)部的BlockingQueue隊列中（即completionQueue）。

4. 獲取任務結(jié)果：

當我們調(diào)用take方法獲取任務結(jié)果時，它會從completionQueue隊列中取出已完成的任務結(jié)果，并返回該結(jié)果。如果隊列為空，則會阻塞等待，直到有任務完成并返回結(jié)果。

take方法內(nèi)部會調(diào)用QueueingFuture的get()方法，從而觸發(fā)對應任務的計算結(jié)果的獲取。

5. 保證按順序獲取結(jié)果：

由于completionQueue是一個阻塞隊列，并且在done()方法中將任務結(jié)果按照完成的順序放入隊列中，因此我們可以通過按順序獲取隊列中的任務結(jié)果，來保證按照任務完成的順序獲取它們的結(jié)果。

通過以上源碼流程步驟，ExecutorCompletionService類能夠按照任務完成的順序獲取結(jié)果。它利用QueueingFuture類包裝任務并存儲結(jié)果到阻塞隊列中，在任務執(zhí)行完成后，按照完成的順序?qū)⒔Y(jié)果放入隊列，從而實現(xiàn)了按順序獲取結(jié)果的功能。

注意事項

在使用ExecutorCompletionService時，需要注意以下事項：

• 合理選擇線程池大?。焊鶕?jù)任務的數(shù)量和復雜性，合理選擇線程池的大小，以充分利用系統(tǒng)資源并避免資源浪費。
• 及時處理異常：在任務執(zhí)行過程中，如果發(fā)生異常，需要及時處理和記錄異常信息，以保證程序的穩(wěn)定性和可靠性。
• 使用Future對象進行任務取消和超時控制：通過使用Future對象的cancel方法，可以取消正在執(zhí)行的任務。同時，可以通過調(diào)整 poll 方法的參數(shù)來設置超時時間，避免長時間等待任務結(jié)果而導致阻塞。

總結(jié)

ExecutorCompletionService是一個強大且靈活的工具類，能夠簡化異步任務的處理和結(jié)果獲取過程。通過使用ExecutorCompletionService，我們可以更加高效地處理一組異步任務，并按照完成的順序獲取它們的結(jié)果。

本文介紹了ExecutorCompletionService的基本使用方法，并對其源碼進行了解析。希望通過這篇博客能夠幫助讀者更好地理解和應用ExecutorCompletionService。