FutureTask

重要的说明

1. FutureTask可以用来包装一个Callable和Runnable对象。FutureTask实现了Runable，所有可以提交给Executor执行。
2. 改变基于AQS来实现阻塞等待的原因，是避免在线程池和线程复用的情况下出现取消竞争过程中保留了中断的状态。
3. private volatile int state状态变化可能情况：
   3.1. NEW -> COMPLETING -> NORMAL
   3.2. NEW->COMPLETING->EXCEPTION
   3.3. NEW->CANCLELED
   3.4. NEW->INTERRUPTING->INTERRUPTED

源码分析

public class FutureTask<V> implements RunnableFuture<V>{
    //volatile修饰的变量不会被缓存在寄存器或者对其他处理器不可见的地方;JVM 保证了每次读变量都从内存中读
    //volatile只能让被他修饰内容具有可见性，但不能保证它具有原子性
    //禁止指令重排序优化(指令重排序：是指CPU采用了允许将多条指令不按程序规定的顺序分开发送给各相应电路单元处理)
    
    //volatile 的读性能消耗与普通变量几乎相同，但是写操作稍慢，因为它需要在本地代码中插入许多内存屏障指令来保证处理器不发生乱序执行。
    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; //被中断
    
    //执行线程的Callback，执行玩设置为空
    private Callable<V> callable;
    
    //执行后的结果
    /** The result to return or exception to throw from get() */
    private Object outcome; // non-volatile, protected by state reads/writes
    
    //真正执行任务的线程
    //he thread running the callable; CASed during run()
    private volatile Thread runner;
    
      /** Treiber stack of waiting threads */
      //保留了在FutureTask上的等待线程列表
    private volatile WaitNode waiters;
    
    //设计链表的原因是 FutureTask的get在任务处于非完成的状态时需要阻塞等待，
    //如果多个线程进行get操作，显然需要链表或队列来维护这些线程等待线程。
    static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;
        WaitNode() { thread = Thread.currentThread(); }
    }

    
    //构造方法 支持包装Callable 和 Runnable，Runnable也是通过Executors包装为Callable
    public FutureTask(Callable<v> callable){
        
        if(callable == null){
            throw new NullPointerException();
        }
        
        this.callable = callable;
        this.state = NEW;
    }
    
    
    public FutureTask(Runnable runnable,V result){
        this.callable = Executors.callable(runnable,result);
        this.state = NEW;
    }
    
    public boolean isCancelled(){
        
        return state >= CANCELLED ;
    }
    
    public boolean isDone() {
        return state != NEW;
    }
    
    //提交线程池或线程启动start时调用 
    public void run() {
        if (state != NEW ||
            !U.compareAndSwapObject(this, RUNNER, 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) {
                    result = null;
                    ran = false;
                    setException(ex); //异常设置异常结果
                }
                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);
        }
    }

    
   public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            //任务没有执行玩 进入awaitDone
            s = awaitDone(false, 0L);
        return report(s);
    }
    
   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();
        return report(s);
    }
    
    
    private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
        // The code below is very delicate, to achieve these goals:
        // - call nanoTime exactly once for each call to park
        // - if nanos <= 0L, return promptly without allocation or nanoTime
        // - if nanos == Long.MIN_VALUE, don't underflow
        // - if nanos == Long.MAX_VALUE, and nanoTime is non-monotonic
        //   and we suffer a spurious wakeup, we will do no worse than
        //   to park-spin for a while
        long startTime = 0L;    // Special value 0L means not yet parked
        WaitNode q = null;
        boolean queued = false;
        for (;;) {
            int s = state;
            // 根据状态来做判断
            //如果状态大于COMPLETING 说明任务已经技结束
            if (s > COMPLETING) {
                if (q != null)
                    q.thread = null; 
                return s;
            }
            //状态处于中间状态COMPLETING  任务已经结束但是任务执行线程还没来得及给outcome
            else if (s == COMPLETING)
                // We may have already promised (via isDone) that we are done
                // so never return empty-handed or throw InterruptedException
                Thread.yield(); //表示当前调用线程让出执行权
            else if (Thread.interrupted()) {
                //表示中断了 则在等待队列中删除该节点并抛出异常
                removeWaiter(q);
                throw new InterruptedException();
            }
            //当前等待节点为nul
            else if (q == null) {
                //如果设置了超时时间，传入的超时时间又是小于等于0 就返回当前状态判断
                if (timed && nanos <= 0L)
                    return s;
                //初始节点
                q = new WaitNode();
            }
            //将当前节点添加到WAITERS，等待任务完成时被唤醒
            else if (!queued)
                queued = U.compareAndSwapObject(this, WAITERS,
                                                q.next = waiters, q);
            else if (timed) {
                final long parkNanos;
                if (startTime == 0L) { // first time
                    startTime = System.nanoTime();
                    if (startTime == 0L)
                        startTime = 1L;
                    parkNanos = nanos;
                } else {
                    long elapsed = System.nanoTime() - startTime;
                    if (elapsed >= nanos) {
                        removeWaiter(q);
                        return state;
                    }
                    parkNanos = nanos - elapsed;
                }
                // nanoTime may be slow; recheck before parking
                if (state < COMPLETING)
                    // //将当前线程挂起指定时间，时间到了没有被唤醒则会抛异常
                    LockSupport.parkNanos(this, parkNanos);
            }
            else
                 //挂起当前线程
                LockSupport.park(this);
        }
    }

    //获取返回结果
    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);
    }
    
    public boolean cancel(boolean mayInterruptIfRunning) {
        //状态不是NEW 就返回false退出
        //状态是NEW， 就通过CAS修改状态为INTERRUPTING 或 CANCELLED，没修过成功也返回false
        if (!(state == NEW &&
              U.compareAndSwapInt(this, STATE, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;
        try {    // in case call to interrupt throws exception
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    if (t != null)
                        t.interrupt();
                } finally { // final state
                    U.putOrderedInt(this, STATE, INTERRUPTED);
                }
            }
        } finally {
            finishCompletion();
        }
        return true;
    }
    
    private void finishCompletion() {
        // assert state > COMPLETING;
        //遍历等待的线程节点
        for (WaitNode q; (q = waiters) != null;) {
            if (U.compareAndSwapObject(this, WAITERS, q, null)) {
                for (;;) {
                    
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;
                        //唤醒等待的线程,唤醒的线程会各自从awaitDone()方法中的LockSupport.park*()阻塞
                        LockSupport.unpark(t);
                    }
                    WaitNode next = q.next;
                    if (next == null)
                        break;
                    q.next = null; // unlink to help gc
                    q = next;
                }
                break;
            }
        }

        done();

        callable = null;        // to reduce footprint
    }
}