Java线程池AbstractExecutorService原理分析

简介

AbstractExecutorService 抽象类派生自 ExecutorService 接口，然后在其基础上实现了几个实用的方法，这些方法提供给子类进行调用。ExecutorService又是继承Executor接口。接下来我们就来一一分析。

源码解析

1. Executor接口
   我们可以看到 Executor 接口非常简单，就一个 void execute(Runnable command) 方法，代表提交一个任务。

   /* 
    * @since 1.5
    * @author Doug Lea
    */
   public interface Executor {
       void execute(Runnable command);
   }

2. ExecutorServer接口
   这些方法都很好理解，一个简单的线程池主要就是这些功能，能提交任务，能获取结果，能关闭线程池，这也是为什么我们经常用这个接口的原因。

   public interface ExecutorService extends Executor {
   
       /**
        * 关闭线程池，已提交的任务继续执行，不接受继续提交新任务
        * Initiates an orderly shutdown in which previously submitted
        * tasks are executed, but no new tasks will be accepted.
        * Invocation has no additional effect if already shut down.
        *
        * <p>This method does not wait for previously submitted tasks to
        * complete execution.  Use {@link #awaitTermination awaitTermination}
        * to do that.
        *
        * @throws SecurityException if a security manager exists and
        *         shutting down this ExecutorService may manipulate
        *         threads that the caller is not permitted to modify
        *         because it does not hold {@link
        *         java.lang.RuntimePermission}{@code ("modifyThread")},
        *         or the security manager's {@code checkAccess} method
        *         denies access.
        */
       void shutdown();
   
       /**
        * 关闭线程池，尝试停止正在执行的所有任务，不接受继续提交新任务
        * 它和前面的方法相比，加了一个单词“now”，区别在于它会去停止当前正在进行的任务
        * Attempts to stop all actively executing tasks, halts the
        * processing of waiting tasks, and returns a list of the tasks
        * that were awaiting execution.
        *
        * <p>This method does not wait for actively executing tasks to
        * terminate.  Use {@link #awaitTermination awaitTermination} to
        * do that.
        *
        * <p>There are no guarantees beyond best-effort attempts to stop
        * processing actively executing tasks.  For example, typical
        * implementations will cancel via {@link Thread#interrupt}, so any
        * task that fails to respond to interrupts may never terminate.
        *
        * @return list of tasks that never commenced execution
        * @throws SecurityException if a security manager exists and
        *         shutting down this ExecutorService may manipulate
        *         threads that the caller is not permitted to modify
        *         because it does not hold {@link
        *         java.lang.RuntimePermission}{@code ("modifyThread")},
        *         or the security manager's {@code checkAccess} method
        *         denies access.
        */
       List<Runnable> shutdownNow();
   
       /**
        * 线程池是否已关闭
        * Returns {@code true} if this executor has been shut down.
        *
        * @return {@code true} if this executor has been shut down
        */
       boolean isShutdown();
   
       /**
        * 这个方法必须在调用shutdown或shutdownNow方法之后调用才会返回true
        * Returns {@code true} if all tasks have completed following shut down.
        * Note that {@code isTerminated} is never {@code true} unless
        * either {@code shutdown} or {@code shutdownNow} was called first.
        *
        * @return {@code true} if all tasks have completed following shut down
        */
       boolean isTerminated();
   
       /**
        * 等待所有任务完成，并设置超时时间
        * Blocks until all tasks have completed execution after a shutdown
        * request, or the timeout occurs, or the current thread is
        * interrupted, whichever happens first.
        *
        * @param timeout the maximum time to wait
        * @param unit the time unit of the timeout argument
        * @return {@code true} if this executor terminated and
        *         {@code false} if the timeout elapsed before termination
        * @throws InterruptedException if interrupted while waiting
        */
       boolean awaitTermination(long timeout, TimeUnit unit)
           throws InterruptedException;
   
       /**
        * 提交一个 Callable 任务
        * Submits a value-returning task for execution and returns a
        * Future representing the pending results of the task. The
        * Future's {@code get} method will return the task's result upon
        * successful completion.
        *
        * <p>
        * If you would like to immediately block waiting
        * for a task, you can use constructions of the form
        * {@code result = exec.submit(aCallable).get();}
        *
        * <p>Note: The {@link Executors} class includes a set of methods
        * that can convert some other common closure-like objects,
        * for example, {@link java.security.PrivilegedAction} to
        * {@link Callable} form so they can be submitted.
        *
        * @param task the task to submit
        * @param <T> the type of the task's result
        * @return a Future representing pending completion of the task
        * @throws RejectedExecutionException if the task cannot be
        *         scheduled for execution
        * @throws NullPointerException if the task is null
        */
       <T> Future<T> submit(Callable<T> task);
   
       /**
        * 提交一个 Runnable 任务，第二个参数将会放到 Future 中，作为返回值
        * 因为 Runnable 的 run 方法本身并不返回任何东西
        * Submits a Runnable task for execution and returns a Future
        * representing that task. The Future's {@code get} method will
        * return the given result upon successful completion.
        *
        * @param task the task to submit
        * @param result the result to return
        * @param <T> the type of the result
        * @return a Future representing pending completion of the task
        * @throws RejectedExecutionException if the task cannot be
        *         scheduled for execution
        * @throws NullPointerException if the task is null
        */
       <T> Future<T> submit(Runnable task, T result);
   
       /**
        * 提交一个 Runnable 任务
        * Submits a Runnable task for execution and returns a Future
        * representing that task. The Future's {@code get} method will
        * return {@code null} upon <em>successful</em> completion.
        *
        * @param task the task to submit
        * @return a Future representing pending completion of the task
        * @throws RejectedExecutionException if the task cannot be
        *         scheduled for execution
        * @throws NullPointerException if the task is null
        */
       Future<?> submit(Runnable task);
   
       /**
        * 执行所有任务，返回 Future 类型的一个 list
        * Executes the given tasks, returning a list of Futures holding
        * their status and results when all complete.
        * {@link Future#isDone} is {@code true} for each
        * element of the returned list.
        * Note that a <em>completed</em> task could have
        * terminated either normally or by throwing an exception.
        * The results of this method are undefined if the given
        * collection is modified while this operation is in progress.
        *
        * @param tasks the collection of tasks
        * @param <T> the type of the values returned from the tasks
        * @return a list of Futures representing the tasks, in the same
        *         sequential order as produced by the iterator for the
        *         given task list, each of which has completed
        * @throws InterruptedException if interrupted while waiting, in
        *         which case unfinished tasks are cancelled
        * @throws NullPointerException if tasks or any of its elements are {@code null}
        * @throws RejectedExecutionException if any task cannot be
        *         scheduled for execution
        */
       <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
           throws InterruptedException;
   
       /**
        * 也是执行所有任务，但是这里设置了超时时间
        * Executes the given tasks, returning a list of Futures holding
        * their status and results
        * when all complete or the timeout expires, whichever happens first.
        * {@link Future#isDone} is {@code true} for each
        * element of the returned list.
        * Upon return, tasks that have not completed are cancelled.
        * Note that a <em>completed</em> task could have
        * terminated either normally or by throwing an exception.
        * The results of this method are undefined if the given
        * collection is modified while this operation is in progress.
        *
        * @param tasks the collection of tasks
        * @param timeout the maximum time to wait
        * @param unit the time unit of the timeout argument
        * @param <T> the type of the values returned from the tasks
        * @return a list of Futures representing the tasks, in the same
        *         sequential order as produced by the iterator for the
        *         given task list. If the operation did not time out,
        *         each task will have completed. If it did time out, some
        *         of these tasks will not have completed.
        * @throws InterruptedException if interrupted while waiting, in
        *         which case unfinished tasks are cancelled
        * @throws NullPointerException if tasks, any of its elements, or
        *         unit are {@code null}
        * @throws RejectedExecutionException if any task cannot be scheduled
        *         for execution
        */
       <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                     long timeout, TimeUnit unit)
           throws InterruptedException;
   
       /**
        * 只有其中的一个任务结束了，就可以返回，返回执行完的那个任务的结果
        * Executes the given tasks, returning the result
        * of one that has completed successfully (i.e., without throwing
        * an exception), if any do. Upon normal or exceptional return,
        * tasks that have not completed are cancelled.
        * The results of this method are undefined if the given
        * collection is modified while this operation is in progress.
        *
        * @param tasks the collection of tasks
        * @param <T> the type of the values returned from the tasks
        * @return the result returned by one of the tasks
        * @throws InterruptedException if interrupted while waiting
        * @throws NullPointerException if tasks or any element task
        *         subject to execution is {@code null}
        * @throws IllegalArgumentException if tasks is empty
        * @throws ExecutionException if no task successfully completes
        * @throws RejectedExecutionException if tasks cannot be scheduled
        *         for execution
        */
       <T> T invokeAny(Collection<? extends Callable<T>> tasks)
           throws InterruptedException, ExecutionException;
   
       /**
        * 同上一个方法，只有其中的一个任务结束了，就可以返回，返回执行完的那个任务的结果，
        * 不过这个带超时，超过指定的时间，抛出 TimeoutException 异常
        * Executes the given tasks, returning the result
        * of one that has completed successfully (i.e., without throwing
        * an exception), if any do before the given timeout elapses.
        * Upon normal or exceptional return, tasks that have not
        * completed are cancelled.
        * The results of this method are undefined if the given
        * collection is modified while this operation is in progress.
        *
        * @param tasks the collection of tasks
        * @param timeout the maximum time to wait
        * @param unit the time unit of the timeout argument
        * @param <T> the type of the values returned from the tasks
        * @return the result returned by one of the tasks
        * @throws InterruptedException if interrupted while waiting
        * @throws NullPointerException if tasks, or unit, or any element
        *         task subject to execution is {@code null}
        * @throws TimeoutException if the given timeout elapses before
        *         any task successfully completes
        * @throws ExecutionException if no task successfully completes
        * @throws RejectedExecutionException if tasks cannot be scheduled
        *         for execution
        */
       <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                       long timeout, TimeUnit unit)
           throws InterruptedException, ExecutionException, TimeoutException;
   }

3. AbstractExecutorService抽象类
   AbstractExecutorService 抽象类派生自 ExecutorService 接口，然后在其基础上实现了几个实用的方法，这些方法提供给子类进行调用。
   这个抽象类实现了 invokeAny 方法和 invokeAll 方法，这里的两个 newTaskFor 方法也比较有用，用于将任务包装成 FutureTask。定义于最上层接口 Executor中的 void execute(Runnable command) 由于不需要获取结果，不会进行 FutureTask 的包装。

   /**
    * RunnableFuture 是用于获取执行结果的，我们常用它的子类 FutureTask
    * @param runnable
    * @param value
    * @param <T>
    * @return
    */
   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);
   }
   
   /**
    * 提交任务
    * @throws RejectedExecutionException {@inheritDoc}
    * @throws NullPointerException       {@inheritDoc}
    */
   public Future<?> submit(Runnable task) {
       if (task == null) throw new NullPointerException();
       // 1. 将任务包装成 FutureTask
       RunnableFuture<Void> ftask = newTaskFor(task, null);
       // 2. 交给执行器执行，execute 方法由具体的子类来实现
       execute(ftask);
       return ftask;
   }
   
   /**
    * @throws RejectedExecutionException {@inheritDoc}
    * @throws NullPointerException       {@inheritDoc}
    */
   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;
   }
   
   /**
    * @throws RejectedExecutionException {@inheritDoc}
    * @throws NullPointerException       {@inheritDoc}
    */
   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 代表是否设置超时机制，超时时间为第三个参数，
    * 如果 timed 为 true，同时超时了还没有一个线程返回结果，那么抛出 TimeoutException 异常
    * the main mechanics of invokeAny.
    */
   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<Future<T>>(ntasks);
       // ExecutorCompletionService 不是一个真正的执行器，参数 this 才是真正的执行器
       // 它对执行器进行了包装，每个任务结束后，将结果保存到内部的一个 completionQueue 队列中
       // 这也是为什么这个类的名字里面有个 Completion 的原因吧。
       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();
   
           // 首先先提交一个任务，后面的任务到下面的 for 循环一个个提交
           // Start one task for sure; the rest incrementally
           futures.add(ecs.submit(it.next()));
           // 提交了一个任务，所以任务数量减 1
           --ntasks;
           // 正在执行的任务数(提交的时候 +1，任务结束的时候 -1)
           int active = 1;
   
           for (;;) {
               // ecs 上面说了，其内部有一个 completionQueue 用于保存执行完成的结果
               // BlockingQueue 的 poll 方法不阻塞，返回 null 代表队列为空
               Future<T> f = ecs.poll();
               // 为 null，说明刚刚提交的第一个线程还没有执行完成
               // 在前面先提交一个任务，加上这里做一次检查，也是为了提高性能
               if (f == null) {
                   if (ntasks > 0) {
                       --ntasks;
                       futures.add(ecs.submit(it.next()));
                       ++active;
                   }
                   // 这里是 else if，不是 if。这里说明，没有任务了，同时 active 为 0 说明
                   // 任务都执行完成了。其实我也没理解为什么这里做一次 break？
                   else if (active == 0)
                       break;
                   // 这里也是 else if。这里说的是，没有任务了，但是设置了超时时间，这里检测是否超时
                   else if (timed) {
                       // 带等待的 poll 方法
                       f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                       // 如果已经超时，抛出 TimeoutException 异常，这整个方法就结束了
                       if (f == null)
                           throw new TimeoutException();
                       nanos = deadline - System.nanoTime();
                   }
                   // 这里是 else。说明，没有任务需要提交，但是池中的任务没有完成，还没有超时(如果设置了超时)
                   // take() 方法会阻塞，直到有元素返回，说明有任务结束了
                   else
                       f = ecs.take();
               }
               // 有任务结束了
               if (f != null) {
                   --active;
                   try {
                       // 返回执行结果，如果有异常，都包装成 ExecutionException
                       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));
   }
   
   // 执行所有的任务，返回任务结果。
   // 先不要看这个方法，我们先想想，其实我们自己提交任务到线程池，也是想要线程池执行所有的任务
   // 只不过，我们是每次 submit 一个任务，这里以一个集合作为参数提交
   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) {
               // 包装成 FutureTask
               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 {
                       // 这是一个阻塞方法，直到获取到值，或抛出了异常
                       // 这里有个小细节，其实 get 方法签名上是会抛出 InterruptedException 的
                       // 可是这里没有进行处理，而是抛给外层去了。此异常发生于还没执行完的任务被取消了
                       f.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);
       }
   }
   
   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 {
                       // 调用带超时的 get 方法，这里的参数 nanos 是剩余的时间，
                       // 因为上面其实已经用掉了一些时间了
                       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);
       }
   }

   到这里，我们发现，这个抽象类包装了一些基本的方法，可是像 submit、invokeAny、invokeAll 等方法，它们都没有真正开启线程来执行任务，它们都只是在方法内部调用了 execute 方法。至于execute是怎么实现的，可以查看另一篇文章 Java线程池ThreadPoolExecutor原理分析。