正文
CompletionService异步非阻塞获取并行任务执行结果
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
第1部分 问题引入
《Java并发编程实践》一书6.3.5节CompletionService：Executor和BlockingQueue，有这样一段话：
"如果向Executor提交了一组计算任务，并且希望在计算完成后获得结果，那么可以保留与每个任务关联的Future，然后反复使用get方法，同时将参数timeout指定为0，从而通过轮询来判断任务是否完成。这种方法虽然可行，但却有些繁琐。幸运的是，还有一种更好的方法：完成服务CompletionService。"
这是什么意思呢？通过一个例子，分别使用繁琐的做法和CompletionService来完成，清晰的对比能让我们更好的理解上面的一段话和CompletionService这个API提供的初衷。
第2部分 实例
考虑这样的场景，有5个Callable任务分别返回5个整数，然后我们在main方法中按照各个任务完成的先后顺序，在控制台打印返回结果。
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);
return returnValue;
}}
1.轮询的做法
通过一个List来保存每个任务返回的Future，然后轮询这些Future，直到每个Future都已完成。我们不希望出现因为排在前面的任务阻塞导致后面先完成的任务的结果没有及时获取的情况，所以在调用get方式时，需要将超时时间设置为0。
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 -1;
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();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
//任务已经完成
if(result!=null){
System.out.println("result = "+result);
//从future列表中删除已经完成的任务
futureList.remove(future);
taskSize --;
break;
}
}
}
// 所有任务已经完成,关闭线程池
System.out.println("all over ");
executor.shutdown();
}
}
执行结果：
2.使用CompletionService
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));
try {
System.out.println("result:"+completionService.take().get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
System.out.println("all over. ");
executor.shutdown(); }
}
执行结果：
第3部分 源码分析
首先看一下 构造方法：
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>>();
}
构造法方法主要初始化了一个阻塞队列，用来存储已完成的task任务。 ExecutorCompletionService是CompletionService的实现，融合了线程池Executor和阻塞队列BlockingQueue的功能。可以推测，按照任务的完成顺序获取结果，就是通过阻塞队列实现的，阻塞队列刚好具有这样的性质：阻塞和有序。
然后看一下 completionService.submit 方法：
public Future<V> submit(Callable<V> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task);
//将我们的callable任务包装成QueueingFuture
executor.execute(new QueueingFuture(f));
return f;
}
可以看到，callable任务被包装成QueueingFuture，而 QueueingFuture是 FutureTask的子类，所以最终执行了FutureTask中的run()方法。来看一下该方法：
public void run() {
//判断执行状态，保证callable任务只被运行一次
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
//这里回调我们创建的callable对象中的call方法
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
//处理执行结果
set(result);
}
} finally {
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
可以看到在该 FutureTask 中执行run方法，最终回调自定义的callable中的call方法，执行结束之后，通过 set(result) 处理执行结果：
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//设置执行结果v，并标记线程执行状态为：NORMAL
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
//完成执行，将执行结果添加到队列
finishCompletion();
}
}
继续跟进finishCompletion()方法，在该方法中找到 done()方法：
protected void done() { completionQueue.add(task); }
可以看到该方法只做了一件事情，就是将执行结束的task添加到了队列中，只要队列中有元素，我们调用take()方法时就可以获得执行的结果。
到这里就已经清晰了，异步非阻塞获取执行结果的实现原理其实就是通过队列来实现的，FutureTask将执行结果放到队列中，先进先出，线程执行结束的顺序就是获取结果的顺序。
