理解 Java Future

概述

Future 相关接口关系如下: Java Future Arch

Future

一个 Future 对象表示一次异步计算的结果,声明如下:

public interface Future<V>

其方法可用于获取任务结果、取消任务、检查任务状态等。

get 方法声明为:

V get() throws InterruptedException, ExecutionException;

将返回任务的结果,如果任务还没有完成,那么 get 就会阻塞当前线程(调用 get 方法的线程)直到:

  • 任务完成,返回结果
  • 任务取消,抛出 CancellationException 异常
  • 当前线程被中断,抛出 ExecutionException 异常

get 可以指定等待时间,如果出现超时,将抛出 TimeoutException 异常:

V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;

cancel 尝试取消任务:

boolean cancel(boolean mayInterruptIfRunning);

如果任务已经完成,已经被取消,或者由于某些原因不能被取消,那么方法将返回 false;如果成功取消了任务,那么方法将返回 true。如果任务正在执行,参数 mayInterruptIfRunning 将决定是否尝试通过中断该任务所在线程来终止任务。

如果任务已经完成,isDone 总是返回 true:

boolean isDone();

以下三种情况都认为任务已完成:

  • 正常结束
  • 抛出异常终止
  • 被取消

如果任务在完成前被取消,那么 isCancelled 总是返回 true:

boolean isCancelled();

RunnableFuture

RunnableFuture 接口继承了 Future 和 Runnable,只定义一个 run 方法,负责启动任务的执行:

void run();

FutureTask

FutureTask 实现了 RunnableFuture,可用于封装 Callable 或者 Runnable 对象,主要在 Executor 框架中使用。在 JDK 7 之前,FutureTask 实现依赖于 AQS;而在 JDK 7 后,其实现直接使用 Unsafe 和 LockSupport。

FutureTask 内部使用一个 int 变量来维护任务状态:

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;

新创建的 FutureTask 状态都为 NEW,状态之间转移关系如图: State Transition

状态的改变只允许发生在方法 set,setException 和 cancel 中,其中 COMPLETING 和 INTERRUPTING 都是中间状态,并且任何大于 COMPLETING 的状态都认为任务已完成。

除此之外,一些重要的变量如下:

//The underlying callable; nulled out after running
private Callable<V> callable;

//The result to return or exception to throw from get()
private Object outcome;

//The thread running the callable; CASed during run()
private volatile Thread runner;

//Treiber stack of waiting threads
private volatile WaitNode waiters;

FutureTask 构造方法接受一个 Callable 或者 Runnable 作为参数。如果是 Runnable 对象,会调用 Executors.callable 将其转化成 Callable:

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;       // effect of volatile, ensure visibility of callable
}

public FutureTask(Runnable runnable, V result) {
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;       // effect of volatile, ensure visibility of callable
}

get 返回任务执行结果,如果任务还没有完成,调用 awaitDone 阻塞直到任务完成。如果任务已经完成,调用 report 方法报告执行结果:

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

考虑到可能有多个线程调用 get 等待任务结果,如果任务没有完成,那么需要将这些线程阻塞并加入等待队列,直到任务完成再唤醒它们。队列节点定义基于 Treiber stack

static final class WaitNode {
    volatile Thread thread;
    volatile WaitNode next;
    WaitNode() { thread = Thread.currentThread(); }
}

awaitDone 负责阻塞线程,等待任务执行完成并响应中断。该方法接受一个 boolean 参数 timed,指定是否需要限时等待,和一个 long 参数指定等待时间(纳秒计),定义为:

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

理解 awaitDone 的关键在于理解 WaitNode q 和 boolean queued 都是本地变量。每一个线程执行 awaitDone 时,都会在自己的栈创建 q 和 queued 变量。

在上述基础上,for 循环逻辑如下:

  1. 首先判断当前线程在等待过程中是否被中断。如果是,那么将线程从等待队列中删除,并抛出 InterruptedException 异常
  2. 否则判断 state 是否大于 COMPLETING。如果是,说明任务执行完成,返回状态
  3. 否则判断 state 是否等于 COMPLETING。如果是,说明任务即将完成,不阻塞当前线程,而是调用 Thread.yield 让出当前线程执行权,让任务得以完成
  4. 否则判断 p == null。如果是,那么为当前线程创建一个 WaitNode,暂时不加入等待队列(有可能下一次循环时,任务即将完成或者已经完成,那么就不需要再将节点加入等待队列了)
  5. 否则判断 queued 是否为 false,如果是,说明线程还没有被加入到等待队列,那么将线程加入等待队列(入队方式可以参考 Treiber stack 入栈)
  6. 如果上述条件都不成立,说明需要阻塞线程

run 会启动 Callable 的执行:

public void run() {
    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 {
                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);
    }
}

run 逻辑如下:

  1. 首先判断状态是否等于 NEW,如果不等于 NEW,说明任务可能处于即将完成、执行完成、被取消、出现异常等各种状态,总之不需要再次启动,那么直接返回
  2. 如果状态为 NEW,那么以 CAS 的方式设置任务执行线程 runner 为当前线程,如果设置失败,说明有其它线程竞争成为任务执行者,此时也直接返回
  3. 在开始调用任务之前,还需要再次检查状态(double check 机制)。如果条件成立,那么调用 Callable 的 run 方法执行任务。如果任务正确执行并返回结果,将调用 set 方法将结果赋值给 outcome;如果运行过程中出现异常,那么调用 setException 方法将异常赋值给 outcome
  4. 可能在 set/setException 设置结果之前,有线程调用了 cancel 方法取消任务,这时 set/setException CAS 更新 state 失败,于是返回,在 finally 处理。如果可能存在中断,那么调用 handlePossibleCancellationInterrupt 处理

set 方法定义为:

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
    }
}

set 方法首先尝试使用 CAS 将状态从 NEW 更新为 COMPLETING,如果更新成功,那么设置结果,随后更新状态为 NORMAL,最后调用 finishCompletion 唤醒等待线程。

setException 定义类似于 set:

protected void setException(Throwable t) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        finishCompletion();
    }
}

只不过将 outcome 设置为 Throwable,并且最终将状态设置为 Exceptional。两个方法最后都调用 finishCompletion,唤醒正在等待任务完成的线程:

private void finishCompletion() {
    for (WaitNode q; (q = 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
}

done 方法默认实现什么也没做。FutureTask 子类可以覆盖该方法,调用 callback 或者做 bookkeeping。

任务执行完成后,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);
}

逻辑如下:

  1. 如果状态 NORMAL,那么做强制类型转换并返回结果
  2. 如果状态大于等于 CANCELLED,说明任务被取消或者执行线程被中断,抛出 CancellationException 异常
  3. 否则说明状态为 EXCEPTIONAL,抛出 ExecutionException 异常

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

cancel 逻辑如下:

  1. 如果任务状态不为 NEW,或者状态本来为 NEW,但是 CAS 更新状态失败,那么直接返回 false
  2. 如果 CAS 更新状态成功,根据 mayInterruptIfRunning 判断是否要中断运行线程 runner
  3. 在最后调用 finishCompletion 唤醒等待线程

cancel 和 set/setException 相互配合。如果 cancel 判断状态不等于 NEW 或者 CAS 更新状态失败,说明任务即将完成或者已经完成或者已经取消(INTERRUPTING 和 INTERRUPTED 只能在 cancel 方法中更新),那么这时就不能取消任务,直接返回 false。同时,如果任务还没有执行完成时 cancel 更改了状态,那么 set/setException 直接返回,在 run 方法中最后调用 handlePossibleCancellationInterrupt 处理中断。

private void handlePossibleCancellationInterrupt(int s) {
    // It is possible for our interrupter to stall before getting a
    // chance to interrupt us.  Let's spin-wait patiently.
    if (s == INTERRUPTING)
        while (state == INTERRUPTING) 
            Thread.yield(); // wait out pending interrupt
    }

handlePossibleCancellationInterrupt 等待状态变为 INTERRUPTED 才返回。

参考

  1. Future
  2. RunnableFuture
  3. FutureTask
  4. 深入学习FutureTask

Previous post: 理解 Java volatile 关键字

Next post: 理解 Java Executor 框架