FutureTask在线程池中应用和源码解析

1. 如果不是始状态或者 cas 设置运行线程是当前线程不成功，直接返回，防止多个线程重复执行。
2. 执行 Callable 的 call()，即提交执行任务（这里做了catch，会捕获执行任务的异常封装到 outcome 中）。
3. 如果成功执行 set 方法，封装结果。

set 方法

protected void set(V v) {
    //cas方式设置成completing状态，防止多个线程同时处理
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v; // 封装结果
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // 最终设置成normal状态

        finishCompletion();
    }
}

解析如下：

1. cas方式设置成completing状态，防止多个线程同时处理
2. 封装结果到outcome，然后设置到最终状态normal
3. 执行finishCompletion方法。

finishCompletion方法

// state > COMPLETING; 不管异常，中断，还是执行完成，都需要执行该方法来唤醒调用get方法阻塞的线程
private void finishCompletion() {
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) {
        // cas 设置waiters为null，防止多个线程执行。
        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
}

解析如下：

遍历waiters中的等待节点，并通过 LockSupport 唤醒每一个节点，通知每个线程，该任务执行完成（可能是执行完成，也可能 cancel，异常等）。

以上就是执行的过程，接下来分析获取结果的过程->get。

get 方法

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        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);
    }

解析如下：

以上两个方法，原理一样，其中一个设置超时时间，支持最多阻塞多长时间。
状态如果小于 COMPLETING，说明还没到最终状态，（不管是否是成功、异常、取消）。
调用 awaitDone 方法阻塞线程，最终调用 report 方法返回结果。

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 (;;) {
            //线程可中断，如果当前阻塞获取结果线程执行interrupt()方法，则从队列中移除该节点，并抛出中断异常
            if (Thread.interrupted()) {
                removeWaiter(q);
                throw new InterruptedException();
            }
            int s = state;
            // 如果已经是最终状态，退出返回
            if (s > COMPLETING) {
                if (q != null)
                    q.thread = null;
                return s;
            }
            //这里做了个优化，competiting到最终状态时间很短，通过yield比挂起响应更快。
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();
            // 初始化该阻塞节点
            else if (q == null)
                q = new WaitNode();
            // cas方式写到阻塞waiters栈中
            else if (!queued)
                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);
        }
    }

解析如下：

整体流程已写到注解中，整体实现是放在一个死循环中，唯一出口，是达到最终状态。
然后是构建节点元素，并将该节点入栈，同时阻塞当前线程等待运行主任务的线程唤醒该节点。

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

然后是report方法，如果是正常结束，返回结果，如果不是正常结束（取消，中断）抛出异常。

最后分析下取消流程。

cancel 方法

public boolean cancel(boolean mayInterruptIfRunning) {
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, 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
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}