07、Java多线程：AbstractExecutorService、CompletionService、ExecutorCompletionService
AbstractExecutorService
public abstract class AbstractExecutorService implements ExecutorService
1、 提供ExecutorService执行方法的默认实现；
2、 此类使用newTaskFor返回的RunnableFuture实现submit、invokeAny和invokeAll方法，默认情况下，RunnableFuture是此包中提供的FutureTask类；
方法
RunnableFuture newTaskFor(Runnable runnable, T value)
为给定可调用任务返回一个 RunnableFuture。
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
RunnableFuture newTaskFor(Callable callable)
为给定可调用任务返回一个 RunnableFuture。
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
FutureTask构造方法
public FutureTask(Runnable runnable, V result) {
sync = new Sync(Executors.callable(runnable, result));
}
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
sync = new Sync(callable);
}
submit(Callable task)
提交一个返回值的任务用于执行，返回一个表示任务的未决结果的 Future。
三种submit方式实现。
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Object> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
List invokeAll(Collection<？ extends Callable> tasks)
执行给定的任务，当所有任务完成时，返回保持任务状态和结果的 Future 列表。
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
boolean done = false;
try {
for (Callable<T> t : tasks) {
RunnableFuture<T> f = newTaskFor(t);
futures.add(f);
execute(f);
}
for (Future<T> f : futures) {
if (!f.isDone()) {
try {
f.get();
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
}
}
}
done = true;
return futures;
} finally {
if (!done)
for (Future<T> f : futures)
f.cancel(true);
}
}
限时等待只是在上述源码中做了超时判断。
invokeAny(Collection<？ extends Callable> tasks)
执行给定的任务，如果某个任务已成功完成（也就是未抛出异常），则返回其结果。
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
try {
return doInvokeAny(tasks, false, 0);
} catch (TimeoutException cannotHappen) {
assert false;
return null;
}
}
/**
* the main mechanics of invokeAny.
*/
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
if (tasks == null)
throw new NullPointerException();
int ntasks = tasks.size();
if (ntasks == 0)
throw new IllegalArgumentException();
List<Future<T>> futures= new ArrayList<Future<T>>(ntasks);
ExecutorCompletionService<T> ecs =
new ExecutorCompletionService<T>(this);
// For efficiency, especially in executors with limited
// parallelism, check to see if previously submitted tasks are
// done before submitting more of them. This interleaving
// plus the exception mechanics account for messiness of main
// loop.
try {
// Record exceptions so that if we fail to obtain any
// result, we can throw the last exception we got.
ExecutionException ee = null;
long lastTime = (timed)? System.nanoTime() : 0;
Iterator<? extends Callable<T>> it = tasks.iterator();
// Start one task for sure; the rest incrementally
futures.add(ecs.submit(it.next()));
--ntasks;
int active = 1;
for (;;) {
Future<T> f = ecs.poll();
if (f == null) {
if (ntasks > 0) {
--ntasks;
futures.add(ecs.submit(it.next()));
++active;
}
else if (active == 0)
break;
else if (timed) {
f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
if (f == null)
throw new TimeoutException();
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
}
else
f = ecs.take();
}
if (f != null) {
--active;
try {
return f.get();
} catch (InterruptedException ie) {
throw ie;
} catch (ExecutionException eex) {
ee = eex;
} catch (RuntimeException rex) {
ee = new ExecutionException(rex);
}
}
}
if (ee == null)
ee = new ExecutionException();
throw ee;
} finally {
for (Future<T> f : futures)
f.cancel(true);
}
}
从源码看出，AbstractExecutorService中submit、invokeAll使用了newTaskFor来实现。
在invokeAny方法中使用了ExecutorCompletionService来判定集合中的任务是否都已执行完成。
CompletionService
public interface CompletionService
1、 将生产新的异步任务与使用已完成任务的结果分离开来的服务；
2、 生产者submit执行的任务使用者take已完成的任务，并按照完成这些任务的顺序处理它们的结果；
通常，CompletionService 依赖于一个单独的 Executor 来实际执行任务，在这种情况下，CompletionService 只管理一个内部完成队列。
方法摘要
Future poll()
获取并移除表示下一个已完成任务的 Future，如果不存在这样的任务，则返回 null。
Future poll(long timeout,TimeUnit unit)throws InterruptedException
获取并移除表示下一个已完成任务的 Future，如果目前不存在这样的任务，则将等待指定的时间（如果有必要）。
Future take() throws InterruptedException
获取并移除表示下一个已完成任务的 Future，如果目前不存在这样的任务，则等待。
Future submit(Runnable task, V result)
提交要执行的 Runnable 任务，并返回一个表示任务完成的 Future，可以提取或轮询此任务。
Future submit(Callable task)
提交要执行的值返回任务，并返回表示挂起的任务结果的 Future。在完成时，可能会提取或轮询此任务。
通过接口定义及解释可看出，CompletionService可以将任务应用与生产t者、消费者模式。
生产者使用 submit 提交需要执行的任务。消费者 take or poll已完成的任务，并可按照任务的完成顺序进行处理。
ExecutorCompletionService
public class ExecutorCompletionService implements CompletionService
1、 使用提供的Executor来执行任务的CompletionService；
2、 此类将安排那些完成时提交的任务，把它们放置在可使用take访问的队列上；
3、 CompletionService的接口实现；
成员变量
private final Executor executor;
private final AbstractExecutorService aes;
private final BlockingQueue<Future<V>> completionQueue;
默认使用LinkedBlockingQueue队列存储已完成的任务队列。
使用executor执行任务。
也可自定义BlockingQueue队列，由其构造方法决定。
ExecutorCompletionService(Executor executor)
or
ExecutorCompletionService(Executor executor,
BlockingQueue<Future<V>> completionQueue)
方法
submit
提交要执行的值返回任务，并返回表示挂起的任务结果的 Future。
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;
}
public Future<V> submit(Runnable task, V result) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task, result);
executor.execute(new QueueingFuture(f));
return f;
}
newTaskFor
private RunnableFuture<V> newTaskFor(Callable<V> task) {
if (aes == null)
return new FutureTask<V>(task);
else
return aes.newTaskFor(task);
}
private RunnableFuture<V> newTaskFor(Runnable task, V result) {
if (aes == null)
return new FutureTask<V>(task, result);
else
return aes.newTaskFor(task, result);
}
内部类QueueingFuture
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;
}
注意其覆写了done()方法，当任务被完成或者取消时，会执行此方法，即将该任务放入已完成队列BlockingQueue中，从而等待被take。
poll
获取并移除表示下一个已完成任务的 Future，如果不存在这样的任务，则返回 null。
public Future<V> poll() {
return completionQueue.poll();
}
public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException {
return completionQueue.poll(timeout, unit);
}
take
获取并移除表示下一个已完成任务的 Future，如果目前不存在这样的任务，则等待。
public Future<V> take() throws InterruptedException {
return completionQueue.take();
}
从源码看出，poll or take即由BlockingQueue实现，那么很容易理解poll直接获取并移除，若没有已完成任务，则可指定等待时间，否则返回null。take与前者的区别在于，当队列中没有已完成任务时，会一直等待，直到有为止。