Java源码阅读之AbstractExecutorService

Summary：

public abstract class AbstractExecutorService implements ExecutorService
实现了submit、invokeAny、invokeAll方法

ExecutorService为止声明的方法有：

void execute(Runnable command);

执行runnable方法

void shutdown();

不再接收新的submit，等待执行完毕，关闭线程

List<Runnable> shutdownNow();

主动去停止当前正在执行的线程

boolean isShutdown();

当前执行器是否还在运行

boolean isTerminated();

所有任务是否已经结束

<T> Future<T> submit(Callable<T> task);

提交一个有返回值的线程任务

<T> Future<T> submit(Runnable task, T result);

提交一个Runnable任务，result为返回值

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)

提交一组任务，当所有任务完成时，返回全部结果；

<T> T invokeAny(Collection<? extends Callable<T>> tasks)

提交一组任务，当任一任务完成时，返回完成的那个结果；

submit(Runnable task)

public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);//方法在ThreadPoolExecutor中实现
return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}

submit(Callable<T> task)

public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}

invokeAny

//提交多个任务，一旦某个任务执行完成，则返回该任务执行结果，其它任务则取消
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;
}
}
//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();
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
ExecutorCompletionService<T> ecs =
new ExecutorCompletionService<T>(this);
try {
ExecutionException ee = null;
final long deadline = timed ? System.nanoTime() + nanos : 0L;
Iterator<? extends Callable<T>> it = tasks.iterator();
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();
nanos = deadline - System.nanoTime();
}
else
f = ecs.take(); //所有任务已经提交，在此等候结果
}
if (f != null) {
--active;
try {
return f.get();
} catch (ExecutionException eex) {
ee = eex;
} catch (RuntimeException rex) {
ee = new ExecutionException(rex);
}
}
}
if (ee == null) //通过break出来的情况
ee = new ExecutionException();
throw ee;
} finally {
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}//取消任务
}

invokeAll

//提交多个任务，直到所有任务执行完毕再返回
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
ArrayList<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 (int i = 0, size = futures.size(); i < size; i++) {
Future<T> f = futures.get(i);
if (!f.isDone()) {  //方法不阻塞
try {
f.get();
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
}
}
}
done = true;
return futures;
} finally {
if (!done)
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}