Java 任务完成队列如何按完成顺序收集结果：异常传播与关闭线程池

批量调用几个外部服务时，按提交顺序等待 Future，往往会让已经完成的任务也排在慢任务后面。Java 的 ExecutorCompletionService 把每个任务完成后的 Future 放进完成队列，调用方可以先消费最先完成的结果；任务内部抛出的异常则在 Future.get() 处重新出现。

要按完成顺序收集结果，就用 ExecutorCompletionService.submit() 提交任务，再循环调用 take() 取完成的 Future；不要把提交时返回的 Future 列表当成完成顺序。

ExecutorCompletionService 解决的是哪一个等待顺序问题

假设三个任务分别耗时 900、120 和 300 毫秒。如果把三个 Future 放进列表后按列表顺序调用 get()，第一个慢任务会挡住后两个已经完成的结果。ExecutorCompletionService 额外维护一个 completionQueue：任务结束时，包装后的 Future 入队，消费者只关心“下一个已经完成的任务”。

它并不会改变线程池的调度策略，也不会让任务本身更快。改变的是结果交付路径：提交顺序保留在任务执行端，消费顺序交给 completionQueue。

最小写法：提交任务后用 take 和 get 收集

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

try {
    for (String url : urls) {
        completionService.submit(() -> fetch(url));
    }

    for (int i = 0; i  future = completionService.take();
        String body = future.get();
        save(body);
    }
} finally {
    executor.shutdown();
}

这里有三条真实调用链：completionService.submit() 把任务交给 executor；任务完成后进入 completionQueue；消费者通过 take() 得到 Future，再由 get() 取出结果。save(body) 的执行顺序就是任务完成顺序。

为什么 take 和 get 必须分开判断

take() 只负责等一个已经完成的 Future，它本身不会告诉你任务是成功还是失败。真正读取结果时，future.get() 可能返回字符串，也可能抛出 ExecutionException。后者表示任务执行体已经结束，但执行体内部抛出了异常；需要通过 getCause() 查看原始原因。

for (int i = 0; i  future = completionService.take();
    try {
        save(future.get());
    } catch (ExecutionException e) {
        logFailure(e.getCause());
    }
}

不要把 InterruptedException 和业务失败混在一起处理。线程被中断时恢复中断标记并退出当前收集流程；业务异常则记录原始 cause，是否继续消费剩余任务由业务决定。

关闭线程池时别让未消费的任务悬在后台

收集循环正常结束后调用 shutdown()，表示不再接收新任务，但已经提交的任务仍会完成。若在收集结果前就关闭并立即依赖进程退出，可能丢失尚未消费的 Future。需要强制停止时才考虑 shutdownNow()，并把返回的未启动任务和中断响应单独记录。

生产代码通常把关闭动作放在 finally，这样 take() 被中断、get() 发生业务异常时也能进入清理路径。若需要等待线程真正结束，再配合 awaitTermination 做有限时长的收口。

三个容易混淆的边界

用可控任务验证完成顺序

验证时不要只打印最终列表，给任务设置不同的短暂等待，并让其中一个任务主动抛出异常。预期现象是：先完成的 Future 先被 take() 取出；失败任务在自己的 get() 位置进入 ExecutionException；所有任务都被消费后才进入 finally 的关闭动作。这样检查的是交付语义，而不是某一次机器上的偶然线程顺序。

相关问题

ExecutorCompletionService 会自动限制并发数吗？

不会。并发上限由传入的 Executor 决定，例如固定线程池的线程数。

take() 和 poll() 怎么选？

take() 会等待完成任务，适合必须收齐结果的循环；poll() 可立即返回 null，适合带超时或周期性检查的消费者。

任务失败后还要继续消费吗？

如果目标是收集每个任务的独立结果，通常继续消费并记录失败；如果一个失败会使整体结果无效，则应取消剩余 Future 并进入清理流程。

小结

ExecutorCompletionService 的价值在于把“任务完成”和“提交顺序”解耦。用 submit() 投递，用 take() 等完成，用 get() 取值并处理异常，最后在 finally 中关闭 executor，就能把并发收集流程写得清楚而可验证。