FutureTask 源码分析

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;

    /** The underlying callable; nulled out after running */
    private Callable<V> callable;
    /** The result to return or exception to throw from get() */
    private Object outcome; // non-volatile, protected by state reads/writes
    /** The thread running the callable; CASed during run() */
    private volatile Thread runner;
    /** Treiber stack of waiting threads */
    private volatile WaitNode waiters;
复制代码

变量 state 记录当前任务的状态，其可取值范围如下：

• NEW : 任务初始状态
• COMPLETING : 任务执行中
• NORMAL : 任务已完成
• CANCELLED : 任务取消
• EXCEPTIONAL : 任务执行异常
• INTERRUPTING : 任务中断
• INTERRUPTED : 任务已中断

状态的流转可能会有以下几种情形：

• NEW -> COMPLETING -> NORMAL
• NEW -> COMPLETING -> EXCEPTIONAL
• NEW -> CANCELLED
• NEW -> INTERRUPTING -> INTERRUPTED

变量 callable 存储的是可执行的任务，变量 outcome 存储任务的返回值，变量 runner 指向当前执行该任务的线程，变量 waiters 执行等待链表的头节点。

初始化

public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();

        this.callable = callable;
        this.state = NEW;
    }
复制代码

FutureTask 在构造时会设置 state 为初始状态 NEW 。

任务的执行 - run()

public void run() {
		// 只有状态 state 为 NEW， runner 为空的情况下才可以执行
        if (state != NEW ||
        	// 将变量 runner 设置为当前线程，
        	// 此处应该是为了保证任务只允许被一个线程处理，也即只允许执行一次
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                	// 执行 callable
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
                	// 成功执行后，设置结果值
                    set(result);
            }
        } finally {        
            runner = null;
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }
复制代码

protected void set(V v) {
		// 设置 state 为 COMPLETING
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        	// 设置返回结果
            outcome = v;
            // 设置 state 为 NORMAL
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            // 任务完成唤醒挂起的线程
            finishCompletion();
        }
    }
复制代码

private void finishCompletion() {
        // assert state > COMPLETING;
        for (WaitNode q; (q = waiters) != null;) {
        	// 将 waiters 重置为空
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            	// 采用死循环的方式唤醒挂起的线程
                for (;;) {
                	// 获取等待节点关联的线程
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;
                        // 唤醒线程
                        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
    }
复制代码

从上述源码中可以看出 callable 只会被执行一次，执行过程如下：

• 设置 runner 为当前线程
• 回调 callable
• 设置状态 state 为 COMPLETING
• 设置返回结果 outcome
• 设置状态 state 为 NORMAL
• 唤醒等待链表 waiters 里的线程

那么 waiters 等待链表什么时候存在等待的节点呢 ？

获取任务结果 - get()

public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    }
复制代码

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();
            else if (q == null)
            	// 创建等待节点
                q = new WaitNode();
            else if (!queued)
            	// 若未加入等待链表时，将 q 的 next 指向 waiters , 然后将 waiters 移动到 q
                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
            	// 挂起调用者线程
            	// 当任务执行完 执行 finishCompletion 是会被唤醒
                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);
    }
复制代码

从 get() 操作源码可以看出，当调用者线程执行 futureTask.get() 时首先判断当前 state 是否大于 COMPLETING; 若小于等于 COMPLETING 则调用 awaitDone 方法； 在 awaitDone 方法中采用轮询的方式执行以下逻辑：

• 创建 WaitNode
• 将调用者线程 WaitNode 的 next 指向 waiters， 然后将 waiters 指向调用者线程 WaitNode
• 若需要超时等待，则将调用者线程按指定时间挂起，反之将调用者线程挂起等待任务完成唤醒
• state > COMPLETING 任务完成退出循环

结合上述分析可得 FutureTask 执行活动图如下：

同时也可以看出，在 FutureTask 中内部维护了一个单向链表 waiters , 在执行 get 的时候会向其中添加节点:

任务取消 - cancel()

/**
	 * mayInterruptIfRunning 是否中断执行线程
	 */
	public boolean cancel(boolean mayInterruptIfRunning) {
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        	// 只有 FutureTask 为初始状态的时候 允许取消
            return false;
        try {    // in case call to interrupt throws exception
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    if (t != null)
                    	// 中断执行线程
                        t.interrupt();
                } finally { // final state
                	// 设置状态为 INTERRUPTED
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
        	// 唤醒挂起的线程
            finishCompletion();
        }
        return true;
    }
复制代码

小结

从上可以看出 FutureTask 可以用于当一个线程需要等待另外一个线程执行完某个任务后才能继续执行的场景下。