“既生 ExecutorService, 何生 CompletionService？”-有解無憂

前言

在我會手動創建執行緒，為什么要使用執行緒池? 中詳細的介紹了 ExecutorService，可以將整塊任務拆分做簡單的并行處理；

在不會用Java Future，我懷疑你泡茶沒我快中又詳細的介紹了 Future 的使用，填補了 Runnable 不能獲取執行緒執行結果的空缺

將二者結合起來使用看似要一招吃天下了（Java有并發，并發之大，一口吃不下）, but ～～是我太天真

ExecutorService VS CompletionService

假設我們有 4 個任務(A, B, C, D)用來執行復雜的計算，每個任務的執行時間隨著輸入引數的不同而不同，如果將任務提交到 ExecutorService，相信你已經可以“信手拈來”

ExecutorService executorService = Executors.newFixedThreadPool(4);
List<Future> futures = new ArrayList<Future<Integer>>();
futures.add(executorService.submit(A));
futures.add(executorService.submit(B));
futures.add(executorService.submit(C));
futures.add(executorService.submit(D));

// 遍歷 Future list，通過 get() 方法獲取每個 future 結果
for (Future future:futures) {
    Integer result = future.get();
    // 其他業務邏輯
}

先直入主題，用 CompletionService 實作同樣的場景

ExecutorService executorService = Executors.newFixedThreadPool(4);

// ExecutorCompletionService 是 CompletionService 唯一實作類
CompletionService executorCompletionService= new ExecutorCompletionService<>(executorService );

List<Future> futures = new ArrayList<Future<Integer>>();
futures.add(executorCompletionService.submit(A));
futures.add(executorCompletionService.submit(B));
futures.add(executorCompletionService.submit(C));
futures.add(executorCompletionService.submit(D));

// 遍歷 Future list，通過 get() 方法獲取每個 future 結果
for (int i=0; i<futures.size(); i++) {
    Integer result = executorCompletionService.take().get();
    // 其他業務邏輯
}

兩種方式在代碼實作上幾乎一毛一樣，我們曾經說過 JDK 中不會重復造輪子，如果要造一個新輪子，必定是原有的輪子在某些場景的使用上有致命缺陷

既然新輪子出來了，二者到底有啥不同呢？在搞定 CompletableFuture，并發異步編程和撰寫串行程式還有什么區別？文中，我們提到了 Future get() 方法的致命缺陷:

如果 Future 結果沒有完成，呼叫 get() 方法，程式會阻塞在那里，直至獲取回傳結果

先來看第一種實作方式，假設任務 A 由于引數原因，執行時間相對任務 B,C,D 都要長很多，但是按照程式的執行順序，程式在 get() 任務 A 的執行結果會阻塞在那里，導致任務 B,C,D 的后續任務沒辦法執行，又因為每個任務執行時間是不固定的，所以無論怎樣調整將任務放到 List 的順序，都不合適，這就是致命弊端

新輪子自然要解決這個問題，它的設計理念就是哪個任務先執行完成，get() 方法就會獲取到相應的任務結果，這么做的好處是什么呢？來看個圖你就瞬間理解了

兩張圖一對比，執行時長高下立判了，在當今高并發的時代，這點時間差，在吞吐量上起到的效果可能不是一點半點了

那 CompletionService 是怎么做到獲取最先執行完的任務結果的呢？

遠看CompletionService 輪廓

如果你使用過訊息佇列，你應該秒懂我要說什么了，CompletionService 實作原理很簡單

就是一個將異步任務的生產和任務完成結果的消費解耦的服務

用人話解釋一下上面的抽象概念我只能再畫一張圖了

說白了，哪個任務執行的完，就直接將執行結果放到佇列中，這樣消費者拿到的結果自然就是最早拿到的那個了

從上圖中看到，有任務，有結果佇列，那 CompletionService 自然也要圍繞著幾個關鍵字做文章了

既然是異步任務，那自然可能用到 Runnable 或 Callable
既然能獲取到結果，自然也會用到 Future 了

帶著這些線索，我們走進 CompletionService 原始碼看一看

近看 CompletionService 原始碼

CompletionService 是一個介面，它簡單的只有 5 個方法：

Future<V> submit(Callable<V> task);
Future<V> submit(Runnable task, V result);
Future<V> take() throws InterruptedException;
Future<V> poll();
Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;

關于 2 個 submit 方法，我在不會用Java Future，我懷疑你泡茶沒我快文章中做了非常詳細的分析以及案例使用說明，這里不再過多贅述

另外 3 個方法都是從阻塞佇列中獲取并移除阻塞佇列第一個元素，只不過他們的功能略有不同

Take: 如果佇列為空，那么呼叫 take() 方法的執行緒會被阻塞
Poll: 如果佇列為空，那么呼叫 poll() 方法的執行緒會回傳 null
Poll-timeout: 以超時的方式獲取并移除阻塞佇列中的第一個元素，如果超時時間到，佇列還是空，那么該方法會回傳 null

所以說，按大類劃分上面5個方法，其實就是兩個功能

提交異步任務（submit）
從佇列中拿取并移除第一個元素 (take/poll)

CompletionService 只是介面，ExecutorCompletionService 是該介面的唯一實作類

ExecutorCompletionService 原始碼分析

先來看一下類結構, 實作類里面并沒有多少內容

ExecutorCompletionService 有兩種建構式：

private final Executor executor;
private final AbstractExecutorService aes;
private final BlockingQueue<Future<V>> completionQueue;

public ExecutorCompletionService(Executor executor) {
    if (executor == null)
        throw new NullPointerException();
    this.executor = executor;
    this.aes = (executor instanceof AbstractExecutorService) ?
        (AbstractExecutorService) executor : null;
    this.completionQueue = new LinkedBlockingQueue<Future<V>>();
}

public ExecutorCompletionService(Executor executor,
                                 BlockingQueue<Future<V>> completionQueue) {
    if (executor == null || completionQueue == null)
        throw new NullPointerException();
    this.executor = executor;
    this.aes = (executor instanceof AbstractExecutorService) ?
        (AbstractExecutorService) executor : null;
    this.completionQueue = completionQueue;
}

兩個建構式都需要傳入一個 Executor 執行緒池，因為是處理異步任務的，我們是不被允許手動創建執行緒的，所以這里要使用執行緒池也就很好理解了

另外一個引數是 BlockingQueue，如果不傳該引數，就會默認佇列為 LinkedBlockingQueue，任務執行結果就是加入到這個阻塞佇列中的

所以要徹底理解 ExecutorCompletionService ，我們只需要知道一個問題的答案就可以了：

它是如何將異步任務結果放到這個阻塞佇列中的？

想知道這個問題的答案，那只需要看它提交任務之后都做了些什么？

public Future<V> submit(Callable<V> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<V> f = newTaskFor(task);
    executor.execute(new QueueingFuture(f));
    return f;
}

我們前面也分析過，execute 是提交 Runnable 型別的任務，本身得不到回傳值，但又可以將執行結果放到阻塞佇列里面，所以肯定是在 QueueingFuture 里面做了文章

從上圖中看一看出，QueueingFuture 實作的介面非常多，所以說也就具備了相應的介面能力，

重中之重是，它繼承了 FutureTask ，FutureTask 重寫了 Runnable 的 run() 方法 (方法細節分析可以查看FutureTask原始碼分析 ) 文中詳細說明了，無論是set() 正常結果，還是setException() 結果，都會呼叫 finishCompletion() 方法:

private void finishCompletion() {
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) {
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    LockSupport.unpark(t);
                }
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                q = next;
            }
            break;
        }
    }

  	// 重點 重點 重點
    done();

    callable = null;        // to reduce footprint
}

上述方法會執行 done() 方法，而 QueueingFuture 恰巧重寫了 FutureTask 的 done() 方法：

方法實作很簡單，就是將 task 放到阻塞佇列中

protected void done() { 
  completionQueue.add(task); 
}

執行到此的 task 已經是前序步驟 set 過結果的 task，所以就可以通過消費阻塞佇列獲取相應的結果了

相信到這里，CompletionService 在你面前應該沒什么秘密可言了

CompletionService 的主要用途

在 JDK docs 上明確給了兩個例子來說明 CompletionService 的用途：

假設你有一組針對某個問題的solvers，每個都回傳一個型別為Result的值，并且想要并發地運行它們，處理每個回傳一個非空值的結果，在某些方法使用(Result r)

其實就是文中開頭的使用方式

 void solve(Executor e,
            Collection<Callable<Result>> solvers)
     throws InterruptedException, ExecutionException {
     CompletionService<Result> ecs
         = new ExecutorCompletionService<Result>(e);
     for (Callable<Result> s : solvers)
         ecs.submit(s);
     int n = solvers.size();
     for (int i = 0; i < n; ++i) {
         Result r = ecs.take().get();
         if (r != null)
             use(r);
     }
 }

假設你想使用任務集的第一個非空結果，忽略任何遇到例外的任務，并在第一個任務準備好時取消所有其他任務

void solve(Executor e,
            Collection<Callable<Result>> solvers)
     throws InterruptedException {
     CompletionService<Result> ecs
         = new ExecutorCompletionService<Result>(e);
     int n = solvers.size();
     List<Future<Result>> futures
         = new ArrayList<Future<Result>>(n);
     Result result = null;
     try {
         for (Callable<Result> s : solvers)
             futures.add(ecs.submit(s));
         for (int i = 0; i < n; ++i) {
             try {
                 Result r = ecs.take().get();
                 if (r != null) {
                     result = r;
                     break;
                 }
             } catch (ExecutionException ignore) {}
         }
     }
     finally {
         for (Future<Result> f : futures)
           	// 注意這里的引數給的是 true，詳解同樣在前序 Future 原始碼分析文章中
             f.cancel(true);
     }

     if (result != null)
         use(result);
 }

這兩種方式都是非常經典的 CompletionService 使用范式，請大家仔細品味每一行代碼的用意

范式沒有說明 Executor 的使用，使用 ExecutorCompletionService，需要自己創建執行緒池，看上去雖然有些麻煩，但好處是你可以讓多個 ExecutorCompletionService 的執行緒池隔離，這種隔離性能避免幾個特別耗時的任務拖垮整個應用的風險（這也是我們反復說過多次的，不要所有業務共用一個執行緒池）

總結

CompletionService 的應用場景還是非常多的，比如

Dubbo 中的 Forking Cluster
多倉庫檔案/鏡像下載（從最近的服務中心下載后終止其他下載程序）
多服務呼叫（天氣預報服務，最先獲取到的結果）

CompletionService 不但能滿足獲取最快結果，還能起到一定 "load balancer" 作用，獲取可用服務的結果，使用也非常簡單，只需要遵循范式即可

并發系列講了這么多，分析原始碼的程序也碰到各種佇列，接下來我們就看看那些讓人眼花繚亂的佇列

靈魂追問

通常處理結果還會用異步方式進行處理，如果采用這種方式，有哪些注意事項？
如果是你，你會選擇使用無界佇列嗎？為什么？
個人博客：https://dayarch.top
加我微信好友, 進群娛樂學習交流，備注「進群」

歡迎持續關注公眾號：「日拱一兵」

前沿 Java 技術干貨分享

高效工具匯總 | 回復「工具」

面試問題分析與解答

技術資料領取 | 回復「資料」

以讀偵探小說思維輕松趣味學習 Java 技術堆疊相關知識，本著將復雜問題簡單化，抽象問題具體化和圖形化原則逐步分解技術問題，技術持續更新，請持續關注......

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

標籤：Java

上一篇：也來說說redis+lua實作高并發限流

下一篇：怎么寫，簡歷才不會被丟到非洲