线程池原理--执行器AbstractExecutorService

本文主要是介绍线程池原理--执行器AbstractExecutorService，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

线程池原理--执行器AbstractExecutorService
- AbstractExecutorService
- Sumit
- 批量提交任务

线程池原理–总索引

线程池原理–执行器AbstractExecutorService

AbstractExecutorService

AbstractExecutorService是一个抽象类，实现自ExecutorService接口。该类实现了接口中的一些方法。

Sumit

可以看到，不管提交的是Runnable类型或者Callable类型的任务，最终都是创建FutureTask类型对象，并提交给execute()方法执行，execute()方法由子类实现。

 public Future<?> submit(Runnable task) {if (task == null) throw new NullPointerException();RunnableFuture<Void> ftask = newTaskFor(task, null);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;}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(Runnable runnable, T value) {return new FutureTask<T>(runnable, value);}protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {return new FutureTask<T>(callable);}

批量提交任务

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<>(tasks.size());try {// 循环提交任务给execute()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);//在这里阻塞一直到f的任务执行完毕。if (!f.isDone()) {try { f.get(); }catch (CancellationException ignore) {}catch (ExecutionException ignore) {}}}return futures;} catch (Throwable t) {cancelAll(futures);throw t;}}

doInvokeAny也是用于批量提交任务，也会发生阻塞，但是只要有一个任务执行完毕，那么便会退出阻塞状态并返回。
并且如果发生任何异常，所有的任务都会被取消。

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<>(ntasks);ExecutorCompletionService<T> ecs =new ExecutorCompletionService<T>(this);// For efficiency, especially in executors with limited// parallelism, check to see if previously submitted tasks are// done before submitting more of them. This interleaving// plus the exception mechanics account for messiness of main// loop.try {// Record exceptions so that if we fail to obtain any// result, we can throw the last exception we got.ExecutionException ee = null;final long deadline = timed ? System.nanoTime() + nanos : 0L;Iterator<? extends Callable<T>> it = tasks.iterator();// Start one task for sure; the rest incrementally//提交第一个任务futures.add(ecs.submit(it.next()));//每提交第一个任务，ntasks-1--ntasks;int active = 1;for (;;) {//检索并移除下一个要完成的任务Future<T> f = ecs.poll();//f == null ，说明还没有任务要完成，继续添加任务执行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, NANOSECONDS);if (f == null)throw new TimeoutException();nanos = deadline - System.nanoTime();}elsef = ecs.take();}//f！= null ，说明已经有任务完成，返回结果if (f != null) {--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 {cancelAll(futures);}}