AbstractExecutorService定义与实现
在讲解ThreadPoolExecutor时说明过继承的层级其中就有AbstractExecutorService，今天将来详细介绍。
//此类的定义并没有特殊的意义仅仅是实现了ExecutorService接口
//而ExecutorService接口的定义在netty结构第一篇有讲述如果不清楚的读者可以去查看一下
public abstract class AbstractExecutorService implements ExecutorService {
//此方法很简单就是对runnable保证，将其包装为一个FutureTask，FutureTask的解读在前文也讲解过
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
//包装callable为FutureTask
//之前说过FutureTask其实就是对Callable的一个封装
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
//提交一个Runnable类型的任务
public Future<?> submit(Runnable task) {
//如果为null则抛出NPE
if (task == null) throw new NullPointerException();
//包装任务为一个Future
RunnableFuture<Void> ftask = newTaskFor(task, null);
//将任务丢给执行器，而此处会抛出拒绝异常，在讲述ThreadPoolExecutor的时候有讲述，不记得的读者可以去再看看
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;
}
//与上方方法相同只是换成了callable
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
//执行集合tasks结果是最后一个执行结束的任务结果
//可以设置超时 timed为true并且nanos是未来的一个时间
//任何一个任务完成都将会返回结果
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
//传入的任务集合不能为null
if (tasks == null)
throw new NullPointerException();
//传入的任务数不能是0
int ntasks = tasks.size();
if (ntasks == 0)
throw new IllegalArgumentException();
//满足上面的校验后将任务分装到一个ArrayList中
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
//并且创建一个执行器传入this
//这里简单讲述他的执行原理，传入this会使用传入的this(类型为Executor)作为执行器用于执行任务，当submit提交任务的时候回将任务
//封装为一个内部的Future并且重写他的done而此方法就是在future完成的时候调用的，而他的写法则是将当前完成的future添加到esc
//维护的结果队列中
ExecutorCompletionService<T> ecs =
new ExecutorCompletionService<T>(this);
try {
//创建一个执行异常，以便后面抛出
ExecutionException ee = null;
//如果开启了超时则计算死线时间如果时间是0则代表没有开启执行超时
final long deadline = timed ? System.nanoTime() + nanos : 0L;
//获取任务的迭代器
Iterator<? extends Callable<T>> it = tasks.iterator();
//先获取迭代器中的第一个任务提交给前面创建的ecs执行器
futures.add(ecs.submit(it.next()));
//前面记录的任务数减一
--ntasks;
//当前激活数为1
int active = 1;
//进入死循环
for (;;) {
//获取刚才提价的任务是否完成如果完成则f不是null否则为null
Future<T> f = ecs.poll();
//如果为null则代表任务还在继续
if (f == null) {
//如果当前任务大于0 说明除了刚才的任务还有别的任务存在
if (ntasks > 0) {
//则任务数减一
--ntasks;
//并且再次提交新的任务
futures.add(ecs.submit(it.next()));
//当前的存活的执行任务加一
++active;
}
//如果当前存活任务数是0则代表没有任务在执行了从而跳出循环
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();
}
//如果获取到了任务结果f!=null
if (f != null) {
//激活数减一
--active;
try {
//返回获取到的结果
return f.get();
//如果获取结果出错则包装异常
} catch (ExecutionException eex) {
ee = eex;
} catch (RuntimeException rex) {
ee = new ExecutionException(rex);
}
}
}
//如果异常不是null则抛出如果是则创建一个
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));
}
//相对于上一个方法执行成功任何一个则返回结果而此方法是全部执行完然后统一返回结果
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
//传入的任务集合不能是null
if (tasks == null)
throw new NullPointerException();
//创建一个集合用来保存获取到的执行future
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
//任务是否执行完成
boolean done = false;
try {
//遍历传入的任务并且调用执行方法将创建的future添加到集合中
for (Callable<T> t : tasks) {
RunnableFuture<T> f = newTaskFor(t);
futures.add(f);
execute(f);
}
//遍历获取到的future
for (int i = 0, size = futures.size(); i < size; i++) {
Future<T> f = futures.get(i);
//如果当前任务没有成功则进行f.get方法等待此方法执行成功，如果方法执行异常或者被取消将忽略异常
if (!f.isDone()) {
try {
f.get();
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
}
}
}
//到这一步则代表所有的任务都已经有了确切的结果
done = true;
//返回任务结果集合
return futures;
} finally {
//如果不是true是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));
//计算最终计算的确切时间点，运行时长不能超过此时间也就是时间死线
//这里是个细节future创建的时间并没有算作执行时间
final long deadline = System.nanoTime() + nanos;
//获取当前结果数
final int size = futures.size();
//遍历将任务进行执行
for (int i = 0; i < size; i++) {
execute((Runnable)futures.get(i));
//并且每次都计算死线
nanos = deadline - System.nanoTime();
//如果时间已经超过则返回结果
if (nanos <= 0L)
return futures;
}
//否则遍历future确定每次执行都获取到了结果
for (int i = 0; i < size; i++) {
Future<T> f = futures.get(i);
if (!f.isDone()) {
//如果在等待过程中已经超时则返回当前等待结合
if (nanos <= 0L)
return futures;
try {
//如果没有超过死线则设置从future中获取结果的时间如果超过则会派出timeout
f.get(nanos, TimeUnit.NANOSECONDS);
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
} catch (TimeoutException toe) {
//抛出了异常则会返回当前的列表
return futures;
}
//计算最新的超时时间
nanos = deadline - System.nanoTime();
}
}
//之前的返回都没有设置为true所以在finally中都会设置为取消唯独正常执行完成到此处返回的结果才是最终的结果
done = true;
return futures;
} finally {
if (!done)
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
}
网友评论