public interface CompletionService<V> { Future<V> submit(Callable<V> task); Future<V> submit(Runnable task, V result); /** * Retrieves and removes the Future representing the next * completed task, waiting if none are yet present. * * @return the Future representing the next completed task * @throws InterruptedException if interrupted while waiting * 阻塞/ Future<V> take() throws InterruptedException; /** * Retrieves and removes the Future representing the next * completed task or <tt>null</tt> if none are present. * * @return the Future representing the next completed task, or * <tt>null</tt> if none are present * 非阻塞/ Future<V> poll(); }
CompletionService 也不是到處都能用，它不適合處理任務數量有限但個數不可知的場景。例如，要統計某個文件夾中的文件個數，在遍歷子文件夾的時候也會“遞歸地”提交新的任務，但最后到底提交了多少，以及在什么時候提交完了所有任務，都是未知數，無論 CompletionService 還是線程池都無法進行判斷。這種情況只能直接用線程池來處理。
CompletionService 接口的實例可以充當生產者和消費者的中間處理引擎，從而達到將提交任務和處理結果的代碼進行解耦的目的。生產者調用submit 方法提交任務，而消費者調用 poll（非阻塞）或 take（阻塞）方法獲取下一個結果：這一特征看起來和阻塞隊列（BlockingQueue）類似，兩者的區別在於 CompletionService 要負責任務的處理，而阻塞隊列則不會。
在 JDK 中，該接口只有一個實現類 ExecutorCompletionService，該類使用創建時提供的 Executor 對象（通常是線程池）來執行任務，然后將結果放入一個阻塞隊列中。
1. 背景
在Java5的多線程中，可以使用Callable接口來實現具有返回值的線程。使用線程池的submit方法提交Callable任務，利用submit方法返回的Future存根，調用此存根的get方法來獲取整個線程池中所有任務的運行結果。
方法一：如果是自己寫代碼，應該是自己維護一個Collection保存submit方法返回的Future存根，然后在主線程中遍歷這個Collection並調用Future存根的get()方法取到線程的返回值。
方法二：使用CompletionService類，它整合了Executor和BlockingQueue的功能。你可以將Callable任務提交給它去執行，然后使用類似於隊列中的take方法獲取線程的返回值。
2. 實現代碼
package com.clzhang.sample.thread; import java.util.*; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Callable; import java.util.concurrent.CompletionService; import java.util.concurrent.ExecutorCompletionService; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; public class ThreadPoolTest4 { // 具有返回值的測試線程 class MyThread implements Callable<String> { private String name; public MyThread(String name) { this.name = name; } @Override public String call() { int sleepTime = new Random().nextInt(1000); try { Thread.sleep(sleepTime); } catch (InterruptedException e) { e.printStackTrace(); } // 返回給調用者的值 String str = name + " sleep time:" + sleepTime; System.out.println(name + " finished..."); return str; } } private final int POOL_SIZE = 5; private final int TOTAL_TASK = 20; // 方法一，自己寫集合來實現獲取線程池中任務的返回結果 public void testByQueue() throws Exception { // 創建線程池 ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE); BlockingQueue<Future<String>> queue = new LinkedBlockingQueue<Future<String>>(); // 向里面扔任務 for (int i = 0; i < TOTAL_TASK; i++) { Future<String> future = pool.submit(new MyThread("Thread" + i)); queue.add(future); } // 檢查線程池任務執行結果 for (int i = 0; i < TOTAL_TASK; i++) { System.out.println("method1:" + queue.take().get()); } // 關閉線程池 pool.shutdown(); } // 方法二，通過CompletionService來實現獲取線程池中任務的返回結果 public void testByCompetion() throws Exception { // 創建線程池 ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE); CompletionService<String> cService = new ExecutorCompletionService<String>(pool); // 向里面扔任務 for (int i = 0; i < TOTAL_TASK; i++) { cService.submit(new MyThread("Thread" + i)); }
// 檢查線程池任務執行結果 for (int i = 0; i < TOTAL_TASK; i++) { Future<String> future = cService.take(); System.out.println("method2:" + future.get()); } // 關閉線程池 pool.shutdown(); } public static void main(String[] args) throws Exception { ThreadPoolTest4 t = new ThreadPoolTest4(); t.testByQueue(); t.testByCompetion(); } }
部分輸出：
... Thread4 finished... method1:Thread4 sleep time:833 method1:Thread5 sleep time:158 Thread6 finished... method1:Thread6 sleep time:826 method1:Thread7 sleep time:185 Thread9 finished... Thread8 finished... method1:Thread8 sleep time:929 method1:Thread9 sleep time:575 ... Thread11 finished... method2:Thread11 sleep time:952 Thread18 finished... method2:Thread18 sleep time:793 Thread19 finished... method2:Thread19 sleep time:763 Thread16 finished... method2:Thread16 sleep time:990 ...
3. 總結
使用方法一，自己創建一個集合來保存Future存根並循環調用其返回結果的時候，主線程並不能保證首先獲得的是最先完成任務的線程返回值。它只是按加入線程池的順序返回。因為take方法是阻塞方法，后面的任務完成了，前面的任務卻沒有完成，主程序就那樣等待在那兒，只到前面的完成了，它才知道原來后面的也完成了。使用方法二，使用CompletionService來維護處理線程的返回結果時，主線程總是能夠拿到最先完成的任務的返回值，而不管它們加入線程池的順序。