高并发编程：ExecutorCompletionService 深入解析

2010 年 7 月 17 日

要点解说

假设现在有一大批需要进行计算的任务，为了提高整批任务的执行效率，你可能会使用线程池，向线程池中不断submit异步计算任务，同时你需要保留与每个任务关联的Future，最后遍历这些 Future，通过调用 Future接口实现类的 get方法获取整批计算任务的各个结果。

虽然使用了线程池提高了整体的执行效率，但遍历这些Future，调用 Future接口实现类的get方法是阻塞的，也就是和当前这个 Future关联的计算任务真正执行完成的时候，get方法才返回结果，如果当前计算任务没有执行完成，而有其它 Future关联的计算任务已经执行完成了，就会白白浪费很多等待的时间，所以最好是遍历的时候谁先执行完成就先获取哪个结果，这样就节省了很多持续等待的时间。

而ExecutorCompletionService可以实现这样的效果，它的内部有一个先进先出的阻塞队列，用于保存已经执行完成的Future，通过调用它的take方法或poll方法可以获取到一个已经执行完成的 Future，进而通过调用 Future接口实现类的get方法获取最终的结果。

实例演示

@Test
public void test() throws InterruptedException, ExecutionException {
Executor executor = Executors.newFixedThreadPool(3);
CompletionService service = new ExecutorCompletionService(executor);
for (int i = 0 ; i < 5 ;i++) {
int seqNo = i;
service.submit(new Callable() {
@Override
public String call() throws Exception {
return “HelloWorld-“ + seqNo + “-“ + Thread.currentThread().getName();
}
});
}
for (int j = 0 ; j < 5; j++) {
System.out.println(service.take().get());
}
}

执行结果：

HelloWorld-2-pool-1-thread-3
HelloWorld-1-pool-1-thread-2
HelloWorld-3-pool-1-thread-2
HelloWorld-4-pool-1-thread-3

HelloWorld-0-pool-1-thread-1

方法解析

ExecutorCompletionService实现了CompletionService接口，在CompletionService接口中定义了如下这些方法：

Future submit(Callable task):提交一个Callable类型任务，并返回该任务执行结果关联的Future；
Future submit(Runnable task,V result):提交一个Runnable类型任务，并返回该任务执行结果关联的Future；
Future take():从内部阻塞队列中获取并移除第一个执行完成的任务，阻塞，直到有任务完成；
Future poll():从内部阻塞队列中获取并移除第一个执行完成的任务，获取不到则返回null，不阻塞；
Future poll(long timeout, TimeUnit unit):从内部阻塞队列中获取并移除第一个执行完成的任务，阻塞时间为timeout，获取不到则返回null；

源码解析

根据上面的实例演示代码分析 ExecutorCompletionService内部的实现原理。

ExecutorCompletionService有三个私有属性，分别是 executor、aes和completionQueue，其中completionQueue就是存储已完成任务的队列，具体代码如下图。

进入它的构造方法，在方法内部给它的三个属性赋值，可以看到在这里初始化了一个LinkedBlockingQueue类型的先进先出阻塞队列，具体代码如下图。

接着，进入ExecutorCompletionService的submit方法，这里我们分析参数类型是Callable的 submit方法，具体代码如下图。

跟踪代码进入newTaskFor方法，具体代码如下图。

在 ExecutorCompletionService构造方法中已经给aes赋过值了，所以进入AbstractExecutorService的newTaskFor方法，具体代码如下图。

跟踪代码进入FutureTask构造方法，具体代码如下图。

到这里构建的RunnableFuture实例对象完成了，回到上述的submit方法中，继续分析executor.execute(new QueueingFuture(f))，首先是 new QueueingFuture(f)，QueueingFuture是 ExecutorCompletionService中的内部类，具体代码如下图。

从图中的代码可以看到，将 RunnableFuture实例对象赋值给了 QueueingFuture的task属性，注意上图红框中有一个done方法，它的内部是将一个task添加到已完成阻塞队列中，这个先记住后面会用到。接着，分析 executor.execute(new QueueingFuture(f))，因为我们的实例演示代码中使用到的是ThreadPoolExecutor，所以 executor.execute()方法执行到 ThreadPoolExecutor中，具体重点代码如下图。