1. 接口CompletionService
接口CompletionService设计目标在于将生产者和消费者解耦，使生产者和消费者异步执行。
这里所说生产者负责提交任务（task）运行产生结果，消费者异步获得运行结果。
该接口定义的一组方法：
// submit提交可执行任务，返回Future用于异步获取结果
1. Future<V> submit(Callable<V> task);
2. Future<V> submit(Runnable task, V result);
//take返回一个完成的任务的结果，没有会一直阻塞。
3. Future<V> take() throws InterruptedException;
// poll返回一个完成的任务的结果，没有返回null
4. Future<V> poll();
// 返回一个完成的任务的结果，没有完成的任务会阻塞timeout时间再返回
5. Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
2. 实现类ExecutorCompletionService
ExecutorCompletionService实现接口CompletionService，下面是ExecutorCompletionService的用法：
public class CompletionServiceTest {
static class RandomSleep implements Callable<Integer>{
private static Random random = new Random(System.currentTimeMillis());
// 实现Callable， 随机sleep n秒，然后返回n。
@Override
public Integer call() throws Exception {
int sleepTime = random.nextInt(20);
System.out.println("sleep " + sleepTime + "s");
Thread.sleep(sleepTime * 1000);
return sleepTime;
}
}
public static void main(String[] args){
// 创建线程池，ExecutorCompletionService本身需要接受传入的线程池来运行任务。
ExecutorService cachedThreadPoll = Executors.newCachedThreadPool();
((ThreadPoolExecutor)cachedThreadPoll).setCorePoolSize(5);
// 基于线程池cachedThreadPoll创建ExecutorCompletionService实例
CompletionService completionService = new ExecutorCompletionService(cachedThreadPoll);
//提交5个任务运行
for(int i = 0; i < 5; ++i){
completionService.submit(new RandomSleep());
}
cachedThreadPoll.shutdown();
for(int i = 0; i < 5; ++i){
try {
// 上面1提到take会返回一个运行成功的任务, 否则会阻塞
System.out.println("get result: " + completionService.take().get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
}
-------------------------------
输出：
sleep 6s
sleep 8s
sleep 9s
sleep 19s
sleep 10s
get result: 6
get result: 8
get result: 9
get result: 10
get result: 19
从上面的输出可以看出take按任务运行结束先后顺序返回。
2.1 基本原理
ExecutorCompletionService有三个成员:
// executor用来执行任务
private final Executor executor;
// 如果executor实现了抽象类AbstractExecutorService，aes就是executor，否则为null
private final AbstractExecutorService aes;
// 任务完成后返回的结果（Future）放入阻塞对列，take，poll即从这个对列取任务运行 结果
private final BlockingQueue<Future<V>> completionQueue;
每一个通过ExecutorCompletionService#submit提交的任务（实现Callable或者Runnable）会被ExecutorCompletionService#newTaskFor包装成QueueingFuture。
QueueingFuture是ExecutorCompletionService内部类，继承关系如下(<<I>>表示接口):
Future<<I>> Runnable<<I>>
^ ^
| |
----------------------
|
RunnableFuture<<I>>
^
|
FutureTask
^
|
QueueingFuture
QueueingFuture继承类FutureTask，FutureTask构造函数包装了一个Callable或则Runnable任务实例，
FutureTask还有一个空的protected方法done()，会在其包装的任务运行成功、任务取消或则任务异常的情况下被调用。
ExecutorConpletionService内部类QueueingFuture继承FutureTask并重写done方法，done方法里将被包装的task放入阻塞队列completionQueue，因此调用ExecutorCompletionService#take或则poll总是会获得运行完成的任务。
注： ExecutorConpletionService#take总是获得完成的任务，但是这个任务可能是正常完成的，也可能是被取消或则抛出了异常。因此还需要对take返回的Future作出判断再进一步处理。
