摘要：抽象類，實現(xiàn)了的接口。將任務(wù)封裝成提交任務(wù)主要方法在任務(wù)是否超時超時時間任務(wù)書用于存放結(jié)果的，先完成的放前面。
AbstractExecutorService抽象類，實現(xiàn)了ExecutorService的接口。
newTaskFor
將任務(wù)封裝成FutureTask
protected RunnableFuture newTaskFor(Runnable runnable, T value) {
return new FutureTask(runnable, value);
}
protected RunnableFuture newTaskFor(Callable callable) {
return new FutureTask(callable);
}
submit
提交任務(wù)
public Future submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public Future submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public Future submit(Callable task) {
if (task == null) throw new NullPointerException();
RunnableFuture ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
invokeAny
主要方法在doInvokeAny
//tasks任務(wù)
//timed是否超時
//nanos超時時間
private T doInvokeAny(Collection> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
if (tasks == null)
throw new NullPointerException();
int ntasks = tasks.size();//任務(wù)書
if (ntasks == 0)
throw new IllegalArgumentException();
ArrayList> futures = new ArrayList>(ntasks);
//用于存放結(jié)果的，先完成的放前面。所以第一個任務(wù)沒完成的時候，會繼續(xù)提交后續(xù)任務(wù)
ExecutorCompletionService ecs =
new ExecutorCompletionService(this);
try {
//異常信息
ExecutionException ee = null;
//過期時間
final long deadline = timed ? System.nanoTime() + nanos : 0L;
Iterator> it = tasks.iterator();//獲取第一個任務(wù)
提交任務(wù)
futures.add(ecs.submit(it.next()));
--ntasks;//因為提交了一個，任務(wù)數(shù)-1
int active = 1;//正在執(zhí)行的任務(wù)
for (;;) {
Future f = ecs.poll();
if (f == null) {//第一個沒完成
if (ntasks > 0) {//還有沒提交的任務(wù)
--ntasks;//任務(wù)數(shù)-1
futures.add(ecs.submit(it.next()));//提交任務(wù)
++active;//正在執(zhí)行的任務(wù)+1
}
else if (active == 0)//當前沒任務(wù)了，但是都失敗了，異常被捕獲了
break;
else if (timed) {
f = ecs.poll(nanos, TimeUnit.NANOSECONDS);//等待
if (f == null)//返回空，超時拋出異常，結(jié)束
throw new TimeoutException();
nanos = deadline - System.nanoTime();//剩余時間
}
else
f = ecs.take();//阻塞等待獲取
}
if (f != null) {//說明已經(jīng)執(zhí)行完
--active;//任務(wù)數(shù)-1
try {
return f.get();//返回執(zhí)行結(jié)果
} catch (ExecutionException eex) {
ee = eex;
} catch (RuntimeException rex) {
ee = new ExecutionException(rex);
}
}
}
if (ee == null)
ee = new ExecutionException();
throw ee;
} finally {
//取消其他任務(wù)，畢竟第一個結(jié)果已經(jīng)返回了
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
public T invokeAny(Collection> tasks)
throws InterruptedException, ExecutionException {
try {
return doInvokeAny(tasks, false, 0);
} catch (TimeoutException cannotHappen) {
assert false;
return null;
}
}
public T invokeAny(Collection> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return doInvokeAny(tasks, true, unit.toNanos(timeout));
}
invokeAll
返回所有任務(wù)的結(jié)果
public List> invokeAll(Collection> tasks)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
ArrayList> futures = new ArrayList>(tasks.size());//
boolean done = false;
try {
for (Callable t : tasks) {//封裝任務(wù)，并提交
RunnableFuture f = newTaskFor(t);
futures.add(f);
execute(f);
}
for (int i = 0, size = futures.size(); i < size; i++) {
Future f = futures.get(i);
if (!f.isDone()) {
try {
f.get();//阻塞，等待結(jié)果
} 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);
}
}
public List> invokeAll(Collection> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
long nanos = unit.toNanos(timeout);
ArrayList> futures = new ArrayList>(tasks.size());
boolean done = false;
try {
for (Callable 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;//每個提交都要判斷，超時了返回Future
}
for (int i = 0; i < size; i++) {
Future 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);
}
}
文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請注明本文地址：http://systransis.cn/yun/75708.html
