Java并发包的学习（二）---Executor和ExecutorService的底层实现原理

一.实现原理 ExecutorService的主要实现类是ThreadPoolExecutor，它是基于线程实现的，继承于AbstractExecutorService。AbstractExecutorService是一个抽象类，实现了ExecutorService的部分方法。

1.AbstractExecutorService

AbstractExecutorService提供了submit、invokeAll、invokeAny的默认实现，子类需要实现其他方法。 
除了execute，其他方法都与执行服务的生命周期管理有关，submit/invokeAll/invokeAny最终都会调用execute，execute决定了到底如何执行任务。

submit方法:

public <T> Future<T> submit(Callable<T> task) {
     if (task == null) throw new NullPointerException();
     RunnableFuture<T> ftask = newTaskFor(task);
     execute(ftask);
     return ftask;
 }

调用newTaskFor生成了一个RunnableFuture，RunnableFuture是一个接口，既扩展了Runnable，又扩展了Future，没有定义新的方法。作为Runnable它表示要执行的任务，传递给execute进行执行；作为Future它表示任务执行的异步结果。

2.newTaskFor

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
     return new FutureTask<T>(callable);
 }

3.FutureTask

FutureTask实现了RunnableFuture接口

FutureTask
 public class FutureTask<V> implements RunnableFuture<V> {
     // 表示状态  
     private volatile int state;
     // 刚开始的状态或任务在运行
     private static final int NEW          = 0;
     // 临时状态，任务即将结果，在设置结果
     private static final int COMPLETING   = 1;
     // 任务正常执行完成
     private static final int NORMAL       = 2;
     // 任务执行抛出异常结束
     private static final int EXCEPTIONAL  = 3;
     // 任务被取消
     private static final int CANCELLED    = 4;
     // 任务在被中断
     private static final int INTERRUPTING = 5;
     // 任务被中断
     private static final int INTERRUPTED  = 6;
    // 表示待执行的任务
     private Callable<V> callable;
    // 表示最终的执行结果或异常
     private Object outcome;
    // 表示运行任务的线程
     private volatile Thread runner;
    // 单向链表，表示等待任务执行结果的线程
     private volatile WaitNode waiters;
    public FutureTask(Callable<V> callable) {
         if (callable == null)
             throw new NullPointerException();
         this.callable = callable;
         // 初始化状态
         this.state = NEW;       // ensure visibility of callable
     }
    public FutureTask(Runnable runnable, V result) {
         // 如果传入的是runnable，则转换为callable
         this.callable = Executors.callable(runnable, result);
         this.state = NEW;       // ensure visibility of callable
     }
 }

 任务执行服务会使用一个线程执行FutureTask的run方法public void run() {
     if (state!= NEW ||
         !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                      null, Thread.currentThread()))
         return;
     try {
         Callable<V> c = callable;
         if (c != null && state == NEW) {
             V result;
             boolean ran;
             try {
                 // 调用callable的call方法
                 result = c.call();
                 ran = true;
             } catch (Throwable ex) {
                 // 捕捉异常，异常保存到outcome并调用finishCompletion唤醒所有等待结果的线程
                 result = null;
                 ran = false;
                 setException(ex);
             }
             // 设置结果，保存到outcome并调用finishCompletion唤醒所有等待结果的线程
             if (ran)
                 set(result);
         }
     } finally {
         // runner must be non-null until state is settled to
         // prevent concurrent calls to run()
         runner = null;
         // state must be re-read after nulling runner to prevent
         // leaked interrupts
         int s = state;
         if (s >= INTERRUPTING)
             handlePossibleCancellationInterrupt(s);
     }
 }

三.任务提交者，通过get方法获取结果，

public V get(long timeout, TimeUnit unit)
     throws InterruptedException, ExecutionException, TimeoutException {
     if (unit == null)
         throw new NullPointerException();
     int s = state;
     // 如果任务还未执行完毕就等待
     if (s <= COMPLETING &&
         (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
         throw new TimeoutException();
     // 调用report报告结果,report根据状态返回结果或抛出异常
     return report(s);
 }
private V report(int s) throws ExecutionException {
     Object x = outcome;
     if (s == NORMAL)
         return (V)x;
     if (s >= CANCELLED)
         throw new CancellationException();
     throw new ExecutionException((Throwable)x);
 }