目录
二、ExecutorCompletionService 的核心思想
三、使用 ExecutorCompletionService 的典型步骤
在 Java 并发编程中，高效管理异步任务及其结果至关重要。java.util.concurrent.ExecutorCompletionService 实现了CompletionService接口，正是为简化已完成任务结果处理而设计的高级工具类。本文将深入解析其工作原理、优势及典型应用场景。
一、传统痛点：Future.get() 的阻塞问题
使用 ExecutorService 提交任务时，我们获得 Future 对象。通过 Future.get() 获取结果时，线程会阻塞直到任务完成。当需要按完成顺序处理多个任务时，代码往往陷入复杂的状态检查：
ExecutorService executor = Executors.newFixedThreadPool(5);
List<Future<String>> futures = new ArrayList<>();
// 提交多个任务
for (int i = 0; i < 10; i++) {
futures.add(executor.submit(() -> doLongTask()));
}
// 按提交顺序获取结果（可能阻塞）
for (Future<String> future : futures) {
String result = future.get(); // 若第一个任务未完成，后续已完成任务也无法处理
process(result);
}
二、ExecutorCompletionService 的核心思想
ExecutorCompletionService 通过内部维护一个 BlockingQueue（默认为 LinkedBlockingQueue）存储已完成任务的 Future 对象。其核心优势在于：
按完成顺序消费：无论任务提交顺序如何，总是优先处理最先完成的任务结果。
解耦生产与消费：任务执行与结果处理分离，提升吞吐量。
核心实现原理：
public class ExecutorCompletionService<V> implements CompletionService<V> {
private final Executor executor;
private final AbstractExecutorService aes;
private final BlockingQueue<Future<V>> completionQueue;
public Future<V> take() throws InterruptedException {
return completionQueue.take(); // 阻塞直到有任务完成
}
// 提交任务时封装Future，任务完成自动入队，动作在FutureTask类实现
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;
}
}
三、使用 ExecutorCompletionService 的典型步骤
import java.util.Random;
import java.util.concurrent.*;
public class Test {
public static void main(String[] args) {
// 1. 创建线程池
ExecutorService threadPool = Executors.newFixedThreadPool(5);
// 2. 构建CompletionService（绑定线程池）
CompletionService<String> completionService = new ExecutorCompletionService<>(threadPool);
// 3. 提交多个任务
for (int i = 0; i < 10; i++) {
completionService.submit(() -> {
return doLongTask(); // 模拟耗时操作
});
}
// 4. 按完成顺序获取并处理结果
for (int i = 0; i < 10; i++) {
try {
Future<String> completedFuture = completionService.take(); // 阻塞直到有任务完成
String result = completedFuture.get();
System.out.println("处理结果: " + result);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
// 5. 关闭线程池
threadPool.shutdown();
}
// 模拟耗时操作
public static String doLongTask() throws Exception {
int tag = new Random().nextInt(999999);
Thread.sleep(5);
return String.valueOf(tag);
}
}
四、关键 API 解析
| 方法名 | 行为说明 |
|---|---|
submit(Callable<V> task) |
提交任务，返回 Future 对象 |
take() |
阻塞获取并移除下一个已完成任务的 Future；无完成任务时等待 |
poll() |
获取并移除下一个已完成任务的 Future；立即返回，无任务时返回 null |
poll(long timeout, TimeUnit unit) |
限时等待获取已完成任务，超时返回 null |
五、适用场景分析
| 场景 | 传统方式痛点 | CompletionService 优势 |
|---|---|---|
| 批量下载文件 | 按提交顺序等待，慢任务阻塞整体 | 谁先下完谁先处理，提升用户体验 |
| 并行查询多个数据源 | 响应时间受最慢数据源制约 | 快速展示部分结果，逐步加载 |
| 任务有严格超时要求 | 遍历 Future 列表检查状态复杂 | 直接使用 poll(timeout) 控制超时流程 |
六、与相关工具对比
| 工具 | 特点 | 适用场景 |
|---|---|---|
ExecutorService.invokeAll() |
返回所有 Future 列表，需遍历检查状态 | 需要所有任务完成后统一处理 |
CompletableFuture |
功能更强大（组合、链式调用），但更复杂 | 复杂异步流程控制 |
| ExecutorCompletionService | 轻量级、按完成顺序消费结果 | 流式处理已完成任务 |
ExecutorCompletionService 通过内部完成队列实现了 “任务完成即处理” 的高效模式，尤其适合：
需要按任务完成顺序处理结果的场景
避免快速任务被慢任务阻塞的流式处理
对任务结果进行实时响应的系统
其简洁的 API 设计使得开发人员能够以最小代价提升并发程序的响应速度与资源利用率。在批量异步任务处理中，合理使用该工具可显著优化系统性能。