ExecutorService中定义了两个批量执行任务的方法,invokeAll()和invokeAny(),在批量执行或多选一的业务场景中非常方便。invokeAll()在所有任务都完成(包括成功/被中断/超时)后才会返回,invokeAny()在任意一个任务成功(或ExecutorService被中断/超时)后就会返回。
AbstractExecutorService实现了这两个方法,本文将先后分析invokeAll()和invokeAny()两个方法的源码实现。
invokeAll()
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(); // 无所谓先执行哪个任务的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); } }
8-12行,先将所有任务都提交到线程池(当然,任何ExecutorService均可)中。
严格来说,不是“提交”,而是“执行”。执行可能是同步或异步的,取决于线程池的策略。不过由于我们仅讨论异步情况(同步同理),用“提交”一词更容易理解。下同。
13-22行,for循环的目的是阻塞调用invokeAll的线程,直到所有任务都执行完毕。当然我们也可以使用其他方式实现阻塞,不过这种方式是最简单的:
- 15行如果f.isDone()返回true,则当前任务已结束,继续检查下一个任务;否则,调用f.get()让线程阻塞,直到当前任务结束。
- 17行无所谓先执行哪一个FutureTask实例的get()方法。由于所有任务并发执行,总体阻塞时间取决于于是耗时最长的任务,从而实现了invodeAll的阻塞调用。
- 18-20行没有捕获InterruptedException。如果有任务被中断,主线程将抛出InterruptedException,以响应中断。
最后,为防止在全部任务结束之前过早退出,23行、25-29行相配合,如果done不为true(未执行到40行就退出了)则取消全部任务。
限时版
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); } }
10-11行,先将所有任务封装为FutureTask,添加到futures列表中。
18-23行,每提交一个任务,就立刻判断是否超时。这样的话,如果在任务全部提交到线程池中之前,就已经达到了超时时间,则能够尽快检查出超时,结束提交并退出。
对于限时版,将封装任务与提交任务拆开是必要的。
28-29行,每次在调用限时版f.get()进入阻塞状态之前,先检查是否超时。这里也是希望超时后,能够尽快发现并退出。
其他同不限时版。