FutureTask源码分析

概述

Future表示一个任务的生命周期，是一个可取消的异步运算，可以把它看作是一个异步操作的结果的占位符，它将在未来的某个时刻完成，并提供对结果的访问。
FutureTask为Future提供了基础实现，如获取任务执行结果何取消任务等。如果尚未完成，获取任务执行结果的线程会被阻塞，一旦执行结束，任务就不能被重启或取消。FutureTask常用来封装Clallable和Runnable，也可以作为一个任务提交到线程池中执行。

FutureTask内部维护了一个由volatule修饰的int型变量-state，待变当前任务的运行状态，state有7种状态。

NEW:新建
COMPLETING:完成
NORMAL:正常运行
EXCEPTIONAL：异常退出
CANCELLED：任务取消
INTERRUPTING：线程中断中。
INTERRUPTED:线程已中断

源码分析

继承体系

核心属性

/** 内部持有的callable任务，运行完毕后置为空*/
private Callable<V> callable;
/** 从get()中返回的结果或抛出的异常 */
private Object outcome; // non-volatile, protected by state reads/writes
/** 运行callable的线程*/
private volatile Thread runner;
/** 使用treiber栈保存等待的线程*/
private volatile WaitNode waiters;

重要方法分析

fun方法

public void run() {
//新建任务，CAS替换runner（运行Callable的线程）为当前线程
       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) {
                   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);
       }
   }

运行任务，如果任务状态为NEW状态，则利用CAS修改为当前线程，执行完毕后调用set方法设置执行的结果。

set方法

protected void set(V v) {
//设置为完成态
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        //设置结果
        outcome = v;
        //设置为最终状态
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        finishCompletion();//执行结束，唤醒等待结果的线程
    }
}

set方法主要是修改状态为完成态，获得结果，设置为结束状态。然后调用finishCompletion方法唤醒等待结果的线程。

finishCompletion

该方法的主要作用就是唤醒所有等待结果的线程来取结果了。

private void finishCompletion() {
       // assert state > COMPLETING;
       for (WaitNode q; (q = waiters) != null;) {
       //移除等待的线程
           if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {              
               for (;;) {//自旋遍历等待队列
                   Thread t = q.thread;
                   if (t != null) {
                       q.thread = null;
                       //唤醒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
       callable = null;        // to reduce footprint
   }

get方法

get方法是去获取结果，如果运行还没有结果，那么就会阻塞。

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
    //任务还未完成，到阻塞队列中去等待
        s = awaitDone(false, 0L);
    return report(s);
}

这个地方调用awaitDone方法

`report`方法

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);
   }

awaitDone方法

awaitDone方法用于等待任务完成，或者因为任务中断或超时而终止。返回任务的完成状态。

private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
        //获取任务返回的结果的最后期限
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        boolean queued = false;
        for (;;) {
            if (Thread.interrupted()) {
            //如果线程中断，移除等待
                removeWaiter(q);
                throw new InterruptedException();
            }

            int s = state;
            if (s > COMPLETING) {//如果任务已经完成
                if (q != null)
                //置空等待节点线程
                    q.thread = null;
                return s;
            }
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();//线程让出CPU时间
            else if (q == null)
            //将q设置为等待节点
                q = new WaitNode();
            else if (!queued)
            //cas修改awit
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                     q.next = waiters, q);
            else if (timed) {
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L) {
                //如果已经超时，则移除等待节点
                    removeWaiter(q);
                    return state;
                }
                //阻塞当前线程
                LockSupport.parkNanos(this, nanos);
            }
            else
                LockSupport.park(this);
        }
    }