《Java并发编程实践》一书6.3.5节CompletionService：Executor和BlockingQueue，有这样一段话：
"如果向Executor提交了一组计算任务，并且希望在计算完成后获得结果，那么可以保留与每个任务关联的Future，然后反复使用get方法，同时将参数timeout指定为0，从而通过轮询来判断任务是否完成。这种方法虽然可行，但却有些繁琐。幸运的是，还有一种更好的方法：完成服务CompletionService。"
这是什么意思呢？我们通过一个例子，分别使用繁琐的做法和CompletionService来完成，清晰的对比能让我们更好的理解上面的一段话和CompletionService这个API提供的初衷。考虑这样的场景，有5个Callable任务分别返回5个整数，然后我们在main方法中按照各个任务完成的先后顺序，在控制台打印返回结果。
package net.aty.completeservice;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
public class ReturnAfterSleepCallable implements Callable<Integer>
{
private int sleepSeconds;
private int returnValue;
public ReturnAfterSleepCallable(int sleepSeconds, int returnValue)
{
this.sleepSeconds = sleepSeconds;
this.returnValue = returnValue;
}
@Override
public Integer call() throws Exception
{
System.out.println("begin to execute.");
TimeUnit.SECONDS.sleep(sleepSeconds);
System.out.println("end to execute.");
return returnValue;
}
}
这个任务会接受2个参数，睡眠指定的时间后，返回指定的结果。睡眠时间越短，意味着任务越先执行完成。
package net.aty.completeservice;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public class TraditionalTest
{
public static void main(String[] args)
{
int taskSize = 5;
ExecutorService executor = Executors.newFixedThreadPool(taskSize);
List<Future<Integer>> futureList = new ArrayList<Future<Integer>>();
for (int i = 1; i <= taskSize; i++)
{
int sleep = taskSize - i; // 睡眠时间
int value = i; // 返回结果
// 向线程池提交任务
Future<Integer> future = executor.submit(new ReturnAfterSleepCallable(sleep, value));
// 保留每个任务的Future
futureList.add(future);
}
// 轮询,获取完成任务的返回结果
while(taskSize > 0)
{
for (Future<Integer> future : futureList)
{
Integer result = null;
try
{
result = future.get(0, TimeUnit.SECONDS);
} catch (InterruptedException e)
{
e.printStackTrace();
} catch (ExecutionException e)
{
e.printStackTrace();
} catch (TimeoutException e)
{
// 超时异常需要忽略,因为我们设置了等待时间为0,只要任务没有完成,就会报该异常
}
// 任务已经完成
if(result != null)
{
System.out.println("result=" + result);
// 从future列表中删除已经完成的任务
futureList.remove(future);
taskSize--;
//此处必须break，否则会抛出并发修改异常。（也可以通过将futureList声明为CopyOnWriteArrayList类型解决）
break; // 进行下一次while循环
}
}
}
// 所有任务已经完成,关闭线程池
System.out.println("all over.");
executor.shutdown();
}
}
可见轮询future列表非常的复杂，而且还有很多异常需要处理，TimeOutException异常需要忽略；还要通过双重循环和break，防止遍历集合的过程中，出现并发修改异常。这么多需要考虑的细节，程序员很容易犯错。
package net.aty.completeservice;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class CompletionServiceTest
{
public static void main(String[] args)
{
int taskSize = 5;
ExecutorService executor = Executors.newFixedThreadPool(taskSize);
// 构建完成服务
CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(
executor);
for (int i = 1; i <= taskSize; i++)
{
int sleep = taskSize - i; // 睡眠时间
int value = i; // 返回结果
// 向线程池提交任务
completionService
.submit(new ReturnAfterSleepCallable(sleep, value));
}
// 按照完成顺序,打印结果
for (int i = 0; i < taskSize; i++)
{
try
{
System.out.println(completionService.take().get());
} catch (InterruptedException e)
{
e.printStackTrace();
} catch (ExecutionException e)
{
e.printStackTrace();
}
}
// 所有任务已经完成,关闭线程池
System.out.println("all over.");
executor.shutdown();
}
}
可见使用CompletionService不会有TimeOutExeception的问题，不用遍历future列表，不用担心并发修改异常。
/** * A service that decouples the production of new asynchronous tasks * from the consumption of the results of completed tasks. Producers * <tt>submit</tt> tasks for execution. Consumers <tt>take</tt> * completed tasks and process their results in the order they * complete. */也就是说，CompletionService实现了生产者提交任务和消费者获取结果的解耦，生产者和消费者都不用关心任务的完成顺序，由CompletionService来保证，消费者一定是按照任务完成的先后顺序来获取执行结果。
public ExecutorCompletionService(Executor executor) {
if (executor == null)
throw new NullPointerException();
this.executor = executor;
this.aes = (executor instanceof AbstractExecutorService) ?
(AbstractExecutorService) executor : null;
this.completionQueue = new LinkedBlockingQueue<Future<V>>();
}到这里可以推测，按照任务的完成顺序获取结果，就是通过阻塞队列实现的，阻塞队列刚好具有这样的性质：阻塞和有序。public Future<V> submit(Callable<V> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task);
executor.execute(new QueueingFuture(f));
return f;
}
/**
* FutureTask extension to enqueue upon completion
*/
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;
}这里简单说明下：FutureTask.done()，这个方法默认什么都不做，就是一个回调，当提交的线程池中的任务完成时，会被自动调用。这也就说时候，当任务完成的时候，会自动执行QueueingFuture.done()方法，将返回结果加入到阻塞队列中，加入的顺序就是任务完成的先后顺序。
JDK并发工具包CompletionService和ExecutorCompletionService的好处和使用场景,布布扣,bubuko.com
JDK并发工具包CompletionService和ExecutorCompletionService的好处和使用场景
原文地址：http://blog.csdn.net/aitangyong/article/details/38169383
