ExecutorCompletionService是Java中用于按任务完成顺序获取异步结果的工具,它封装ExecutorService并结合BlockingQueue,将已完成任务的Future放入队列,通过take()或poll()方法获取结果,适用于需优先处理先完成任务的场景,如竞态查询。使用时需创建线程池,提交Callable或Runnable任务,并循环获取结果,注意及时关闭线程池及处理异常。
在Java中,当需要提交多个异步任务并按完成顺序获取结果时,ExecutorCompletionService 是一个非常实用的工具。它结合了 ExecutorService 和 BlockingQueue 的能力,能将已完成的任务结果自动放入队列中,让你可以及时处理最先完成的任务,而不需要等待所有任务结束。
它是一个包装类,接收一个 ExecutorService 实例,并维护一个阻塞队列(通常是 LinkedBlockingQueue),用于存放已完成任务的 Future 对象。每当一个任务完成,其 Future 就会被放入该队列,调用者可以通过 take() 或 poll() 方法获取结果。
相比直接使用 ExecutorService.invokeAll() 等待全部完成,CompletionService 更适合需要“谁先完成就处理谁”的场景,比如竞态查询、批量请求中的快速响应等。
下面是使用 CompletionService 的基本步骤:
示例代码:
import java.util.concurrent.*;public class CompletionServiceExample { public static void main(String[] args) throws InterruptedException, ExecutionException { // 创建线程池 ExecutorService executor = Executors.newFixedThreadPool(4); // 创建CompletionService ExecutorCompletionService
completionService = new ExecutorCompletionService<>(executor); // 提交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() 会阻塞直到有结果可用 Futurefuture = completionService.take(); System.out.println(future.get()); // 输出已完成的任务结果 } executor.shutdown(); } }
输出示例(顺序不固定,取决于执行时间):
任务3 完成,耗时:687ms 任务1 完成,耗时:923ms 任务5 完成,耗时:1102ms 任务2 完成,耗时:1988ms 任务4 完成,耗时:2501mstake() 与 poll() 的区别
CompletionService 提供两种方式从队列中取结果:
例如,使用带超时的 poll:
Futurefuture = completionService.poll(5, TimeUnit.SECONDS); if (future != null) { System.out.println(future.get()); } else { System.out.println("在5秒内没有任务完成"); }
这在某些实时性要求高的场景中很有用。
CompletionService 特别适用于以下情况:
注意点:
基本上就这些。使用 ExecutorCompletionService 能让异步结果的收集更高效、更灵活,特别适合关注响应速度而非执行顺序的并发场景。