FutureTask (JDK1.8)

发表于由

本文基于 JDK 1.8.0_101 对应的源码。

FutureTask 表示一个可取消的异步计算的任务，其构造器接收一个 Runnable/Callable 表示异步计算的任务。虽然说 FutureTask 支持取消，但任务一旦开始执行，执行过程就可能不受取消操作的影响。

当任务没有执行结束时，所有获取结果的操作都会阻塞等待。

以栈的方式维护等待获取结果的线程，因为不用考虑公平性的问题，用栈更简单。

FutureTask 在 JDK 1.7 的时候是基于 AQS 实现的，可以看这里。

总体上：

  • volatile int state 控制执行状态，
  • volatile Thread runner 防止重复执行，
  • volatile WaitNode waiters 以栈形式维护等待线程。

// 要异步执行的任务，执行完后置为 null
private Callable<V> callable;

// 执行的结果或执行中抛出的异常
private Object outcome; // 非 volatile, 通过读写 state 字段来保护

// 执行任务的线程，在 run 方法里通过 CAS 来设置。
// 还可以用来防止重复执行
private volatile Thread runner;

// 维护等待线程的栈
private volatile WaitNode waiters;

// 任务的执行状态
private volatile int state;

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) {
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;       // ensure visibility of callable
}

1. 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;

状态的变化可能有如下的路径：

NEW -> COMPLETING -> NORMAL         // 正常执行完成
NEW -> COMPLETING -> EXCEPTIONAL    // 执行抛出了异常
NEW -> CANCELLED                    // 
NEW -> INTERRUPTING -> INTERRUPTED  //

COMPLETING 状态是为了用来保护设置 outcome 字段的中间状态、非常短暂。

2. 执行任务

执行任务的前提是状态为 NEW、且执行线程为空。

通过对 runner 字段的 CAS 设置来防止被重复执行。

任务不管是正常执行完还是执行过程中抛出了异常，都要唤醒所有等待获取结果的线程。

从下面代码 #1 注释的前后可知，NEW 状态有可能是任务还没开始执行，也可能是执行完成了、但还没更新结果。

public void run() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread())) // 防止重复执行
        return;
    try {
        Callable<V> c = callable;
        if (c != null && state == NEW) { // 状态仍然是 NEW
            V result;
            boolean ran;
            try {
                result = c.call(); // #1
                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 必须保持非空直至 state 设置为终态来防止并发调用 run 方法
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

protected void setException(Throwable t) {
    // 非 NEW 状态，说明被取消了，忽略结果
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        finishCompletion();
    }
}

protected void set(V v) {
    // 非 NEW 状态，说明被取消了，忽略结果
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // 终态
        finishCompletion();             // 唤醒所有等待获取结果的线程
    }
}

2.0 结果处理

返回结果最终是根据状态来决定的。哪怕任务执行完成、但由于被取消成功，也不会返回执行的结果。

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

2.1 唤醒所有等待线程

该方法实现的亮点是：把 waiters 的引用保存到方法本地变量，然后把 waiters 设置为 null，这样就只有这个方法能访问到那个栈，后续的唤醒过程就不用考虑并发问题。

private void finishCompletion() {
    // assert state > COMPLETING;
    // 之所以要循环是可能在释放栈上线程的过程中，又有新的节点加入等待。
    for (WaitNode q; (q = waiters) != null;) {

        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
        // 把 waiters 设为 null，就只有本地变量 q 指向那个栈，可以不用考虑线程安全问题
            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
}

3. 取消

必须是 NEW 状态的才能取消。

取消操作最多是对执行线程 runner 设置中断标志位，如果任务 callable 已经在执行中，是不会终止的，除非在执行过程中判断了中断标志位。

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

4. 获取结果

获取结果时，如果已完成则直接获取，否则可能需要进行等待，其核心是 awaitDone 方法。

awaitDone 方法里对状态的判断顺序是保证正确性的前提：
0. 每次循环都重写读取最新的状态至本地变量，用于后续多次判断。
1. 如果任务已经是终态则返回。
2. 如果任务 COMPLETING 则不需要挂起线程，直接 busy loop 等待最终状态。
3. 否则处于 NEW 状态，通过多次循环完成如下：
　3.1 如果等待节点为空，则新建；
　3.2 如果等待节点还没入栈，则压入栈顶；
　3.3 挂起线程、等待唤醒。

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();
            // 任务已经执行完成、在设置结果的过程中，不需要挂起线程

        // 以下逻辑的都是 NEW 状态下的
        else if (q == null)
            q = new WaitNode();
        else if (!queued) {
            q.next = waiters;
            // 把当前线程节点放入栈顶
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset, waiters, q);
        } else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                removeWaiter(q);
                return state;
            }
            LockSupport.parkNanos(this, nanos);
        }
        else
            LockSupport.park(this);
    }
}

欢迎关注我的微信公众号: coderbee笔记 。

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注

此站点使用Akismet来减少垃圾评论。了解我们如何处理您的评论数据。