提供 ExecutorService 执行方法的默认实现。此类使用 newTaskFor 返回的 RunnableFuture 来实现 commit，invokeAny 和 invokeAll 方法，该方法默认为此包中提供的 FutureTask 类。例如，submit(Runnable) 的实现创建一个关联的 RunnableFuture，该关联的 RunnableFuture 将被执行并返回。子类可以重写 newTaskFor 方法以返回 RunTask 以外的 RunnableFuture 实现。
既然 newTaskFor 这么重要，我们就先来看下 newTaskFor 方法：、
//对 runnable 进行包装，内部会通过 RunnableAdapter 把 runnable 封装成 Callable
//value：执行成功后的默认值
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
//返回给定可调用任务的RunnableFuture
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
看到这，你可能会疑惑 FutureTask 是什么东西？这里我们先了解下 Future 是什么？
答：Future 是一个一部运算的结果。RunnableFuture 其实是 Runable 和 Future 的组合接口。详情请看我的另一篇文章 Future
方法分析
1. submit
/**
* 接收一个 Runable。如果使用 Future.get 方法获取结果的话会返回 null
*/
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
/**
* 接收一个 Runable 和一个默认值。如果使用 Future.get 方法获取结果的话会返回 result 这个默认值
*/
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;
}
/**
* 接收一个 Callable
*/
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
可以看到三个 submit 方法都会先把 task 封装成 RunnableFuture。接着最终都会调用 Executor.execute 方法，所以 AbstractExecutorService 的大部分实现类只要实现 execute 方法就可以（可以看下 ThreadPoolExecutor）。对于 Runable 的 task submit 和 execute 的最大区别是 execute 无法使用 Future.get 方法等待计算完成，纯粹是一个异步计算。
2. invokeAny
// invokeAny 的内部调用方法
// 内部使用了 ExecutorCompletionService
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();
// 用于在第一个task 正常完成或发生异常后，取消其它线程
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();
// 启动一个 task
futures.add(ecs.submit(it.next()));
--ntasks;
int active = 1;
for (;;) {
// 获取队列的 Future，如果队列为空则返回 null。所以如果第一个启动的 task 还未完成，这时候返回的四 null
Future<T> f = ecs.poll();
// 第一个 task 还未完成
if (f == null) {
if (ntasks > 0) {// 如果剩余的任务的数量大于 0，则继续往线程池添加任务
--ntasks;
futures.add(ecs.submit(it.next()));
++active;
}
// 在 return f.get(); 这一步发生异常时 active 才会等于 0。也就是说提交的任务在 return f.get(); 时都发生了异常才会走到这一步
else if (active == 0)
break;
// 1. 所有的任务都提交后
// 2. 没有一个任务完成 || （有任务完成但出现异常 && 有在执行的任务）
// 3. 使用了超时机制
// 满足上面所有条件则进到这一步
else if (timed) {
// 等待获取结果
f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
if (f == null)
throw new TimeoutException();
nanos = deadline - System.nanoTime();
}
// 1. 所有的任务都提交后
// 2. 没有一个任务完成 || （有任务完成但出现异常 && 有在执行的任务）
// 3. 没有使用超时机制
// 满足上面所有条件则进到这一步
else
// take 方法会阻塞直到有结果产生
f = ecs.take();
}
if (f != null) {
// 剩余的任务减 1
--active;
try {
// 获取结果
return f.get();
} catch (ExecutionException eex) {
ee = eex;
} catch (RuntimeException rex) {
ee = new ExecutionException(rex);
}
}
}
// 走到这里说明没有正常获取的结果
if (ee == null)
ee = new ExecutionException();
throw ee;
} finally {
// 获取结果后如果还有任务在执行则取消剩余任务
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
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;
}
}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return doInvokeAny(tasks, true, unit.toNanos(timeout));
}
这里用到了 ExecutorCompletionService ，不熟悉的可以看下我的这篇文章，还有队列的知识。
3. invokeAll
invokeAll 相比较 invokeAny 会简单一点，就是使用了 ArrayList 来存储结果。
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 使用 get 方法进入等待
if (!f.isDone()) {
try {
f.get();
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
}
}
}
done = true;
return futures;
} finally {
// 如果出现异常则 done == false，则把剩余的执行中的线程取消
if (!done)
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
long nanos = unit.toNanos(timeout);
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
boolean done = false;
try {
for (Callable<T> t : tasks)
futures.add(newTaskFor(t));
final long deadline = System.nanoTime() + nanos;
final int size = futures.size();
// Interleave time checks and calls to execute in case
// executor doesn't have any/much parallelism.
for (int i = 0; i < size; i++) {
execute((Runnable)futures.get(i));
nanos = deadline - System.nanoTime();
if (nanos <= 0L)
return futures;
}
for (int i = 0; i < size; i++) {
Future<T> f = futures.get(i);
if (!f.isDone()) {
if (nanos <= 0L)
return futures;
try {
f.get(nanos, TimeUnit.NANOSECONDS);
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
} catch (TimeoutException toe) {
return futures;
}
nanos = deadline - System.nanoTime();
}
}
done = true;
return futures;
} finally {
if (!done)
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
好了 AbstractExecutorService 就到这，下一篇分析 AbstractExecutorService 的一个实现 ThreadPoolExecutor。以上内容有错误的请大家指正。
网友评论