ExecutorCompletionService是Java中用于按任务完成顺序获取异步结果的工具,它封装ExecutorService并结合BlockingQueue,将已完成任务的Future放入队列,通过take()或poll()方法获取结果,适用于需优先处理先完成任务的场景,如竞态查询。使用时需创建线程池,提交Callable或Runnable任务,并循环获取结果,注意及时关闭线程池及处理异常。

在Java中,当需要提交多个异步任务并按完成顺序获取结果时,ExecutorCompletionService 是一个非常实用的工具。它结合了 ExecutorService 和 BlockingQueue 的能力,能将已完成的任务结果自动放入队列中,让你可以及时处理最先完成的任务,而不需要等待所有任务结束。
什么是ExecutorCompletionService?
它是一个包装类,接收一个 ExecutorService 实例,并维护一个阻塞队列(通常是 LinkedBlockingQueue),用于存放已完成任务的 Future 对象。每当一个任务完成,其 Future 就会被放入该队列,调用者可以通过 take() 或 poll() 方法获取结果。
相比直接使用 ExecutorService.invokeAll() 等待全部完成,CompletionService 更适合需要“谁先完成就处理谁”的场景,比如竞态查询、批量请求中的快速响应等。
如何使用CompletionService提交任务并获取结果
下面是使用 CompletionService 的基本步骤:
立即学习“Java免费学习笔记(深入)”;
- 创建线程池(ExecutorService)
- 构建 ExecutorCompletionService 实例,传入线程池
- 通过 submit() 提交多个 Callable 或 Runnable 任务
- 循环调用 take() 或 poll() 获取已完成任务的结果
示例代码:
import java.util.concurrent.*;
<p>public class CompletionServiceExample {
public static void main(String[] args) throws InterruptedException, ExecutionException {
// 创建线程池
ExecutorService executor = Executors.newFixedThreadPool(4);
// 创建CompletionService
ExecutorCompletionService<String> completionService =
new ExecutorCompletionService<>(executor);</p><pre class="brush:php;toolbar:false;"> // 提交5个任务
for (int i = 1; i <= 5; i++) {
final int taskId = i;
completionService.submit(() -> {
// 模拟耗时不同的任务
long sleepTime = (long) (Math.random() * 3000);
Thread.sleep(sleepTime);
return "任务" + taskId + " 完成,耗时:" + sleepTime + "ms";
});
}
// 按完成顺序获取结果
for (int i = 0; i < 5; i++) {
// take() 会阻塞直到有结果可用
Future<String> future = completionService.take();
System.out.println(future.get()); // 输出已完成的任务结果
}
executor.shutdown();
}}
输出示例(顺序不固定,取决于执行时间):
任务3 完成,耗时:687ms 任务1 完成,耗时:923ms 任务5 完成,耗时:1102ms 任务2 完成,耗时:1988ms 任务4 完成,耗时:2501ms
take() 与 poll() 的区别
CompletionService 提供两种方式从队列中取结果:
- take():阻塞方法,如果没有完成的任务,会一直等待直到有结果可用
- poll():非阻塞或带超时的方法,可以设置等待时间,超时返回 null
例如,使用带超时的 poll:
Future<String> future = completionService.poll(5, TimeUnit.SECONDS);
if (future != null) {
System.out.println(future.get());
} else {
System.out.println("在5秒内没有任务完成");
}
这在某些实时性要求高的场景中很有用。
适用场景与注意事项
CompletionService 特别适用于以下情况:
- 多个独立任务执行时间差异大,希望尽快处理已完成的任务
- 实现“搜索竞速”,比如向多个数据源发起请求,取第一个返回的结果
- 需要流式处理任务结果,而不是批量等待
注意点:
- 记得关闭线程池(shutdown),避免资源泄漏
- 处理异常时,务必调用 future.get() 并捕获 ExecutionException
- 如果任务数量不确定,可以用 while 循环配合 poll 判断是否还有任务
基本上就这些。使用 ExecutorCompletionService 能让异步结果的收集更高效、更灵活,特别适合关注响应速度而非执行顺序的并发场景。










