Java 併發編程 79：CompletionService 與 ExecutorCompletionService 異步任務結果有序處理

Java 併發編程 79：CompletionService 與 ExecutorCompletionService 異步任務結果有序處理

場景痛點

主線程提交多個異步任務，希望 「有任務完成就處理結果，並按完成順序逐個處理」。ExecutorService.invokeAll 需等待所有完成，不符合需求。

CompletionService 接口

public interface CompletionService<V> {
    Future<V> submit(Callable<V> task);
    Future<V> submit(Runnable task, V result);
    Future<V> take() throws InterruptedException;     // 阻塞等待下一個完成
    Future<V> poll();                                 // 非阻塞，無完成返回 null
    Future<V> poll(long timeout, TimeUnit unit);     // 限時等待
}

ExecutorCompletionService 實現原理

• 內部持有 Executor 執行任務、BlockingQueue<Future<V>> 存放完成結果
• 提交時包裝為 QueueingFuture (繼承 FutureTask)
• 任務完成 (正常/異常/取消) 觸發 done() → 將 Future 入隊
• take()/poll() 直接從隊列取出

// 核心源碼邏輯
public Future<V> submit(Callable<V> task) {
    RunnableFuture<V> f = newTaskFor(task);
    executor.execute(new QueueingFuture(f)); // 包裝層
    return f;
}

private class QueueingFuture extends FutureTask<Void> {
    QueueingFuture(RunnableFuture<V> task) { super(task, null); this.task = task; }
    protected void done() { completionQueue.add(task); } // 關鍵：完成入隊
    private final Future<V> task;
}

基本使用範例：URL 標題並發解析

public static void parse(List<String> urls) throws InterruptedException {
    ExecutorService executor = Executors.newFixedThreadPool(10);
    CompletionService<String> cs = new ExecutorCompletionService<>(executor);
    
    for (String url : urls) cs.submit(new UrlTitleParser(url));
    
    for (int i = 0; i < urls.size(); i++) {
        Future<String> result = cs.take(); // 阻塞直到有結果
        try { System.out.println(result.get()); }
        catch (ExecutionException e) { e.printStackTrace(); }
    }
    executor.shutdown();
}

模擬 invokeAny：任意一個成功即返回

public static String search(List<String> engines, String keyword) 
        throws InterruptedException {
    ExecutorService executor = Executors.newFixedThreadPool(10);
    CompletionService<String> cs = new ExecutorCompletionService<>(executor);
    List<Future<String>> futures = new ArrayList<>();
    String result = null;
    
    try {
        for (String e : engines) futures.add(cs.submit(new SearchTask(e, keyword)));
        
        for (int i = 0; i < engines.size(); i++) {
            try {
                result = cs.take().get();
                if (result != null) break; // 獲得第一個有效結果
            } catch (ExecutionException ignore) { /* 繼續等待下一個 */ }
        }
    } finally {
        for (Future<String> f : futures) f.cancel(true); // 取消其餘任務
        executor.shutdown();
    }
    return result;
}

關鍵優勢總結

• 解耦「提交」與「結果收集」
• 天然支援「先完成先處理」模式
• 簡單實現 invokeAny、超時控制、部分失敗容忍等進階模式
• 內部隊列預設 LinkedBlockingQueue，可自定義有界隊列防止 OOM

下一節將探討定時任務調度 (ScheduledExecutorService)。

原始碼：github.com/swiftma/program-logic