ExecutorCompletionService 怎样按完成顺序取结果

当一批 Callable 任务并发执行时，最容易写出的代码是保存一组 Future，再按提交顺序逐个调用 get()。这样做并不代表结果按完成顺序消费：排在前面的慢任务会挡住后面已经完成的任务。

更合适的做法是：用 ExecutorCompletionService 提交任务，再从它维护的完成队列中取出 Future。队列返回的是“下一个已经完成的任务”，而不是“第几个提交的任务”。

官方地址：https://docs.oracle.com/en/java/javase/25/docs/api/java.base/java/util/concurrent/ExecutorCompletionService.html

想按完成顺序处理结果，就让 CompletionService 负责收集已完成任务；用 take() 等待结果，用 poll() 做非阻塞或限时检查，最后对取出的 Future 调用 get()。

先理解：它为什么能按完成顺序返回

ExecutorCompletionService 自己不执行线程，它接收一个 Executor，把提交的任务交给执行器运行，并把完成后的 Future 放入完成队列。消费端只关心“下一个完成的结果”，不必再维护提交顺序和等待位置之间的对应关系。

这个模型适合批量远程调用、并行读取和多个候选计算等任务。提交顺序仍然可以由业务代码决定，但结果处理顺序交给完成队列。

最小写法：submit 后循环 take

下面的例子提交三个带返回值的任务，然后循环取出三个完成结果。循环次数按提交任务数确定，这一点很重要：只取一个结果时，剩余已完成任务仍留在队列里。

import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

ExecutorService executor = Executors.newFixedThreadPool(3);
CompletionService completion = new ExecutorCompletionService(executor);

try {
    // 提交顺序只决定任务进入执行器的顺序，不决定结果消费顺序。
    completion.submit(() -> load("商品-1"));
    completion.submit(() -> load("商品-2"));
    completion.submit(() -> load("商品-3"));

    for (int i = 0; i  future = completion.take();
        try {
            // 任务内部异常会在 get 时以 ExecutionException 暴露。
            String result = future.get();
            System.out.println(result);
        } catch (ExecutionException taskFailure) {
            // 单个任务失败时记录并继续消费其余已提交任务。
            System.err.println("任务失败：" + taskFailure.getCause());
        }
    }
} finally {
    // 不再接收新任务后释放线程池资源。
    executor.shutdown();
}

这里的关键不是 Future 的列表，而是 CompletionService 内部的完成队列。每次 take() 拿到一个已完成对象，再通过 get() 读取它的返回值。

take、poll 和 get 怎样配合

三个方法解决的是不同问题：take() 会一直等到有完成任务；无参 poll() 只检查当前队列，没有结果时立即返回 null；带超时的 poll 最多等待指定时长，仍没有结果时返回 null。

// 无结果时立即返回，适合轮询或配合其他工作。
Future ready = completion.poll();

// 最多等待 200 毫秒；超时只表示本次没有取到 Future。
Future maybeReady = completion.poll(200, TimeUnit.MILLISECONDS);
if (maybeReady == null) {
    System.out.println("本轮没有已完成任务");
} else {
    try {
        // poll 的超时与任务本身失败是两种不同边界。
        System.out.println(maybeReady.get());
    } catch (ExecutionException taskFailure) {
        System.err.println("任务执行失败：" + taskFailure.getCause());
    }
}

不要把 poll() 返回 null 当成任务失败，也不要只捕获 InterruptedException 就忽略 ExecutionException。前者是队列暂时没有完成项，后者表示已经取到了 Future，但任务执行阶段抛出了异常。

线程池关闭和几个常见误区

第一，CompletionService 不替你关闭执行器。它使用的是传入的 Executor，因此在所有结果已消费、或明确决定取消剩余任务后，应由调用方关闭 ExecutorService。

第二，按完成顺序不等于按业务顺序。如果页面、账务或批处理必须保持输入顺序，就要给结果保留序号，再在消费端重新排序；完成队列只优化“谁先准备好谁先处理”。

第三，任务数和取结果次数要匹配。提交了五个任务却只循环取两个结果，剩下的完成对象仍可能占用队列；如果只需要第一个成功结果，应在拿到目标结果后取消其余 Future，并在 finally 中完成收尾。

相关问题

ExecutorCompletionService 会改变线程池的并发度吗？

不会。并发度主要由传入的执行器决定，ExecutorCompletionService 负责的是完成结果的收集与消费接口。

为什么拿到 Future 还要调用 get？

完成队列存放的是 Future 对象。取出它表示任务已经完成，但返回值或异常仍要通过 get() 读取。

什么时候不用 CompletionService？

如果任务数量很少、必须严格按输入顺序处理，或者已有一套明确的异步编排模型，直接管理 Future 或使用其他并发抽象可能更直观。需要“谁先完成就先处理”时，再选它更合适。

小结：submit 负责把任务交给执行器，完成队列负责暴露已完成的 Future，take/poll 负责选择等待策略，get 负责读取结果和暴露任务异常。把这四层职责分开，按完成顺序取结果就不会被提交顺序卡住。