在 Java 中，Runnable 接口示意一个没有返回后果的工作，而 Callable 接口示意具备返回后果的工作。
在并发编程中，异步执行提交工作和获取工作后果这两个操作，能够进步零碎的吞吐量。Future 接口应运而生，它示意异步工作的执行后果，并提供了查看工作是否执行完、勾销工作、获取工作执行后果等性能。FutureTask 是 Future 接口的根本实现，常与线程池实现类 ThreadPoolExecutor 配合应用。

本文基于 jdk1.8.0_91

1. 继承体系

RunnableFuture 接口同时实现了 Runnable 接口和 Future 接口，是一种冗余设计。

java.util.concurrent.RunnableFuture

/** * A {@link Future} that is {@link Runnable}. Successful execution of * the {@code run} method causes completion of the {@code Future} * and allows access to its results. *  * @see FutureTask * @see Executor * @since 1.6 * @author Doug Lea * @param <V> The result type returned by this Future's {@code get} method */public interface RunnableFuture<V> extends Runnable, Future<V> {    /**     * Sets this Future to the result of its computation     * unless it has been cancelled.     */    void run();}

FutureTask 是一个可勾销的异步工作，是对 Future 接口的根本实现，具备以下性能：

启动或中断的工作的执行；
判断工作是否执行实现；
获取工作执行实现后的后果。

同时，FutureTask 能够用于包装 Callable 或 Runnable 对象。
因为它实现了 Runnable 接口，能够提交给 Executor 执行。

/** * A cancellable asynchronous computation.  * * @since 1.5 * @author Doug Lea * @param <V> The result type returned by this FutureTask's {@code get} methods */public class FutureTask<V> implements RunnableFuture<V>

java.util.concurrent.Executor

/** * An object that executes submitted {@link Runnable} tasks. * * @since 1.5 * @author Doug Lea */public interface Executor {    void execute(Runnable command);}

2. 属性

java.util.concurrent.FutureTask

// The run state of this task, initially NEW.// 工作的执行状态，初始为 NEW。private volatile int state;/** 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、runner、waiters 三个属性在并发时存在争用，采纳 CAS 保护其准确性。

// Unsafe mechanicsprivate static final sun.misc.Unsafe UNSAFE;private static final long stateOffset;private static final long runnerOffset;private static final long waitersOffset;static {    try {        UNSAFE = sun.misc.Unsafe.getUnsafe();        Class<?> k = FutureTask.class;        stateOffset = UNSAFE.objectFieldOffset            (k.getDeclaredField("state"));        runnerOffset = UNSAFE.objectFieldOffset            (k.getDeclaredField("runner"));        waitersOffset = UNSAFE.objectFieldOffset            (k.getDeclaredField("waiters"));    } catch (Exception e) {        throw new Error(e);    }}

2.1 状态定义

/** * The run state of this task, initially NEW.  The run state * transitions to a terminal state only in methods set, * setException, and cancel.  During completion, state may take on * transient values of COMPLETING (while outcome is being set) or * INTERRUPTING (only while interrupting the runner to satisfy a * cancel(true)). Transitions from these intermediate to final * states use cheaper ordered/lazy writes because values are unique * and cannot be further modified. * * Possible state transitions: * NEW -> COMPLETING -> NORMAL * NEW -> COMPLETING -> EXCEPTIONAL * NEW -> CANCELLED * NEW -> INTERRUPTING -> INTERRUPTED */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 中应用 state 代表工作在运行过程中的状态。随着工作的执行，状态将一直地进行转变。

状态的阐明：

NEW：新建状态，工作都从该状态开始。
COMPLETING：工作正在执行中。
NORMAL：工作失常执行实现。
EXCEPTIONAL：工作执行过程中抛出了异样。
CANCELLED：工作被勾销(不响应中断)。
INTERRUPTING：工作正在被中断。
INTERRUPTED：工作曾经中断。

状态转移过程：

NEW -> COMPLETING -> NORMALNEW -> COMPLETING -> EXCEPTIONALNEW -> CANCELLEDNEW -> INTERRUPTING -> INTERRUPTED

状态的分类：

工作的初始状态：NEW
工作的中间状态：COMPLETING、INTERRUPTING
工作的终止状态：NORMAL、EXCEPTIONAL、CANCELLED、INTERRUPTED

2.2 状态应用

FutureTask 中判断工作是否已勾销、是否已实现，是依据 state 来判断的。

public boolean isCancelled() {    return state >= CANCELLED; // CANCELLED、INTERRUPTING、INTERRUPTED}public boolean isDone() {    return state != NEW;}

能够看到：

被勾销或被中断的工作（CANCELLED、INTERRUPTING、INTERRUPTED），都视为已勾销。
当工作来到了初始状态 NEW，就视为工作已完结。工作的两头态很短暂，并不代表工作正在执行，而是工作曾经执行完了，正在设置最终的返回后果。

依据状态值，FutureTask 能够保障曾经实现的工作不会被再次运行或者被勾销。

中间状态尽管是一个刹时状态，在 FutureTask 中用于线程间的通信。例如：

在 FutureTask#run 中检测到状态 >= INTERRUPTING，阐明其余线程发动了勾销操作，以后线程需期待对方实现中断。
在 FutureTask#get 中检测到状态 <= COMPLETING，阐明执行工作的线程尚未解决完，以后线程需期待对方实现工作。

2.2 栈（Treiber stack）

/** Treiber stack of waiting threads */private volatile WaitNode waiters; // 栈顶指针/** * Simple linked list nodes to record waiting threads in a Treiber * stack.  See other classes such as Phaser and SynchronousQueue * for more detailed explanation. */static final class WaitNode {    volatile Thread thread; // 期待工作执行后果的线程    volatile WaitNode next; // 栈的下一个节点    WaitNode() { thread = Thread.currentThread(); }}

FutureTask 应用链表来结构栈（Treiber stack，应用 CAS 保障栈操作的线程平安，参考 java.util.concurrent.SynchronousQueue.TransferStack）。
其中 waiters 是链表的头节点，代表栈顶的指针。

栈的作用：
FutureTask 实现了 Future 接口，如果获取后果时，工作还没有执行结束，那么获取后果的线程就在栈中挂起，直到工作执行结束被唤醒。

3. 构造函数

赋值工作，设置工作的初始状态。

/** * Creates a {@code FutureTask} that will, upon running, execute the * given {@code Callable}. * * @param  callable the callable task * @throws NullPointerException if the callable is null */public FutureTask(Callable<V> callable) {    if (callable == null)        throw new NullPointerException();    this.callable = callable;    this.state = NEW;       // ensure visibility of callable}/** * Creates a {@code FutureTask} that will, upon running, execute the * given {@code Runnable}, and arrange that {@code get} will return the * given result on successful completion. * * @param runnable the runnable task * @param result the result to return on successful completion. If * you don't need a particular result, consider using * constructions of the form: * {@code Future<?> f = new FutureTask<Void>(runnable, null)} * @throws NullPointerException if the runnable is null */public FutureTask(Runnable runnable, V result) {    this.callable = Executors.callable(runnable, result);    this.state = NEW;       // ensure visibility of callable}

值得注意的两个中央：

FutureTask 创立的时候，状态为 NEW。
因为 FutureTask 应用 Callable 示意工作，需用 Executors#callable 办法将 Runnable 转换为 Callable。

测试：

@Testpublic void executors() throws Exception {    Callable<String> callable = Executors.callable(new Runnable() {        @Override        public void run() {            System.out.println("run!");        }    }, "haha");    String call = callable.call();    System.out.println("call = " + call);}

执行后果：

run!call = haha

4. Runnable 实现

4.1 FutureTask#run

代码流程：

校验工作是否可执行：工作已执行或其余线程已获取执行权，则无奈执行。
调用 Callable#call 执行工作。
若工作执行失败，应用 setException 办法设置异样。
若工作执行胜利，应用 set 办法设置返回后果。
最初，革除对以后线程的记录，判断是否期待中断。

留神，在工作执行完结后，属性 runner、callable 都会被清空。

java.util.concurrent.FutureTask#run

public void run() {    // state != NEW 阐明工作曾经执行结束，不再反复执行    // 将 runner 属性设置为以后线程，若设置失败阐明其余线程已获取执行权    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(); // 执行 Callable#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) // INTERRUPTING、INTERRUPTED            handlePossibleCancellationInterrupt(s);    }}

4.1.1 FutureTask#set

工作执行胜利之后，调用该办法。
用于设置工作状态、设置工作执行后果、唤醒栈中期待工作执行后果的线程。

java.util.concurrent.FutureTask#set

/** * Sets the result of this future to the given value unless * this future has already been set or has been cancelled. * * <p>This method is invoked internally by the {@link #run} method * upon successful completion of the computation. * * @param v the value */protected void set(V v) {    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { // state: NEW -> COMPLETING        outcome = v;        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state: COMPLETING -> NORMAL        finishCompletion();    }}

状态变动：NEW -> COMPLETING -> NORMAL

因为 state 属性是 volatile，这里 putOrderedInt 和 putIntVolatile 是等价的，保障可见性。

为什么这里应用 lazySet 而没有用 CAS ：

在并发状况下，只有一个线程执行 CAS 将 state 从 NEW 批改为 COMPLETING 会胜利，其余线程均失败。
因而随后只有一个线程持续批改 state 为 NORMAL，不存在争用，无需应用 CAS。

4.1.2 FutureTask#setException

工作执行产生异样，调用该办法。
除了设置工作状态不同，其余与 FutureTask#set 雷同。

状态变动：NEW -> COMPLETING -> EXCEPTIONAL

java.util.concurrent.FutureTask#setException

/** * Causes this future to report an {@link ExecutionException} * with the given throwable as its cause, unless this future has * already been set or has been cancelled. * * <p>This method is invoked internally by the {@link #run} method * upon failure of the computation. * * @param t the cause of failure */protected void setException(Throwable t) {    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { // state: NEW -> COMPLETING        outcome = t;        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state: COMPLETING -> EXCEPTIONAL         finishCompletion();    }}

4.1.3 FutureTask#finishCompletion

执行结束，唤醒期待线程。

java.util.concurrent.FutureTask#finishCompletion

/** * Removes and signals all waiting threads, invokes done(), and * nulls out callable. */private void finishCompletion() {    // assert state > COMPLETING;    for (WaitNode q; (q = waiters) != null;) {        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {         // CAS 将 waiters 属性置空：1. CAS 胜利，遍历链表唤醒所有节点；2. CAS 失败，从新读取 waiters            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}

4.1.4 FutureTask#handlePossibleCancellationInterrupt

在 FutureTask#cancel 办法中，会先将 state 设为 INTERRUPTING，再中断 runner 线程，最初将 state 设为 INTERRUPTED。

所以在 FutureTask#run 的 finally 块中如果查看到 state == INTERRUPTING，阐明其余线程发动了 cancel(true) 操作，这里须要期待其余线程中断以后线程。直到检测到 state != INTERRUPTING，阐明其余线程已实现中断以后线程操作。

java.util.concurrent.FutureTask#handlePossibleCancellationInterrupt

/** * Ensures that any interrupt from a possible cancel(true) is only * delivered to a task while in run or runAndReset. */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) // 其余线程中断以后线程之后，会设置 state 为 INTERRUPTED，使这里完结循环            Thread.yield(); // wait out pending interrupt    // assert state == INTERRUPTED;    // We want to clear any interrupt we may have received from    // cancel(true).  However, it is permissible to use interrupts    // as an independent mechanism for a task to communicate with    // its caller, and there is no way to clear only the    // cancellation interrupt.    //    // Thread.interrupted();}

4.2 FutureTask#runAndReset

反对周期性执行工作：

执行工作胜利，不必返回工作后果，也不必扭转工作状态（放弃为 NEW），下次能够再次执行工作。
执行工作失败，则设置异样后果，并批改工作状态（不为 NEW），下次无奈再次执行工作。
勾销执行工作，则期待其余线程中断以后线程，并批改工作状态（不为 NEW），下次无奈再次执行工作。

/** * designed for use with tasks that intrinsically execute more    // 设计用来反对定时工作 * than once. * * @return {@code true} if successfully run and reset */protected boolean runAndReset() {    if (state != NEW ||        !UNSAFE.compareAndSwapObject(this, runnerOffset,                                     null, Thread.currentThread()))        return false;    boolean ran = false;    int s = state;    try {        Callable<V> c = callable;        if (c != null && s == NEW) {            try {                c.call(); // don't set result                ran = true;            } catch (Throwable ex) {                setException(ex); // 批改 state: NEW -> COMPLETING -> EXCEPTIONAL            }        }    } 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        s = state;        if (s >= INTERRUPTING)            handlePossibleCancellationInterrupt(s);    }    return ran && s == NEW; // 返回 true 则容许下次再执行 runAndReset}

5. Future 实现

5.1 Future#get

获取工作执行的后果：

如果工作未实现（NEW、COMPLETING），取后果的线程会阻塞（或自旋）。
如果工作执行出错（EXCEPTIONAL），抛出 ExecutionException
如果工作被勾销了（CANCELLED、INTERRUPTING、INTERRUPTED），抛出 CancellationException
如果线程期待被中断，抛出 InterruptedException

java.util.concurrent.FutureTask#get()

/** * @throws CancellationException {@inheritDoc} */public V get() throws InterruptedException, ExecutionException {    int s = state;    if (s <= COMPLETING)        s = awaitDone(false, 0L); // 自旋或阻塞期待工作实现    return report(s);             // 获取工作执行后果或抛出异样}

5.1.1 FutureTask#awaitDone

期待工作实现（工作执行实现、工作执行出现异常、工作勾销执行），若以后线程产生中断、超时则进行期待。

在自旋中进行判断：

若以后线程已中断，则将节点出栈，抛出 InterruptedException。
若 state > COMPLETING，阐明工作曾经实现，返回以后 state。
若 state == COMPLETING，阐明工作行将实现，以后线程持续自旋。
若 state < COMPLETING，须要将以后线程入栈期待：
- 无超时工夫，始终期待直到被其余线程唤醒（FutureTask#run 或 FutureTask#cancel）或产生中断（Thread#interrupt）；
- 有超时工夫，阻塞直到超时、被唤醒、产生中断。若已超时，将节点出栈，返回 state。

java.util.concurrent.FutureTask#awaitDone

/** * Awaits completion or aborts on interrupt or timeout. * * @param timed true if use timed waits * @param nanos time to wait, if timed * @return state upon completion */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();   // 创立节点，设置 q.thread        else if (!queued)            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,                                                 q.next = waiters, q); // 节点 q 入栈，作为新的头节点 waiters        else if (timed) {            nanos = deadline - System.nanoTime();            if (nanos <= 0L) {                removeWaiter(q);  // 已超时，将节点出栈                return state;            }            LockSupport.parkNanos(this, nanos);        }        else            LockSupport.park(this); // 进入阻塞，由 FutureTask#run 或 FutureTask#cancel 来唤醒（外部均调用 FutureTask#finishCompletion）    }}

5.1.2 FutureTask#report

以后线程期待结束，获取工作的执行后果，或者抛出异样。

java.util.concurrent.FutureTask#report

/** * Returns result or throws exception for completed task. * * @param s completed state value */@SuppressWarnings("unchecked")private V report(int s) throws ExecutionException {    Object x = outcome;    if (s == NORMAL)        return (V)x;    if (s >= CANCELLED) // CANCELLED、INTERRUPTING、INTERRUPTED        throw new CancellationException();    throw new ExecutionException((Throwable)x);}

5.2 Future#get(timeout, unit)

在肯定的工夫之内，期待获取工作执行的后果。

/** * @throws CancellationException {@inheritDoc} */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(); // 期待超时了，工作还没有执行完，则抛出 TimeoutException    return report(s);}

5.3 Future#cancel

尝试勾销工作的执行：

如果工作已实现或已勾销，则勾销操作会失败，返回 false。
如果工作还未执行，则勾销操作会胜利，返回 true。
如果工作正在执行，办法的参数就会批示线程是否须要中断：
- mayInterruptIfRunning 为 true，则以后正在执行的工作会被中断；
- mayInterruptIfRunning 为 false，则容许正在执行的工作持续运行，直到它执行完。

状态变动：
NEW -> CANCELLED
NEW -> INTERRUPTING -> INTERRUPTED

public boolean cancel(boolean mayInterruptIfRunning) {    // 如果工作还没有启动（NEW），则批改工作状态（INTERRUPTING or CANCELLED），批改胜利则进入下一步    // 如果工作状态不是 NEW，则间接返回。阐明工作已完结（已实现、已勾销、出现异常），无奈勾销，返回 false    if (!(state == NEW &&          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))        return false;    try {    // in case call to interrupt throws exception         // 进入这里，阐明工作状态为 INTERRUPTING or CANCELLED        // mayInterruptIfRunning 为 true 阐明须要中断执行工作的线程，为 false 容许工作继续执行完        if (mayInterruptIfRunning) {             try {                Thread t = runner;                if (t != null)                    t.interrupt();            } finally { // final state                // 只有一个线程会执行到这里，无需应用 CAS                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // INTERRUPTING -> INTERRUPTED             }        }    } finally {        finishCompletion(); // 唤醒期待线程    }    return true;}

6. 示例

应用三个线程顺次执行：提交工作、期待工作、勾销工作。
察看执行后果，了解并发状况下多个线程之间如何应用 Future 进行交互。

/** * 三个线程顺次执行：提交工作、期待工作、勾销工作 * 在工作未执行完的时候，勾销工作。 *  * @author Sumkor * @since 2021/4/28 */@Testpublic void cancel() throws InterruptedException {    // 定义工作    FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {        @Override        public String call() throws Exception {            Thread.sleep(10000);            return "哦豁";        }    });    CountDownLatch submitGate = new CountDownLatch(1); // 期待工作提交    CountDownLatch endGate = new CountDownLatch(3);    // 期待线程执行完    // 提交工作    new Thread(new Runnable() {        @Override        public void run() {            try {                submitGate.countDown();                System.out.println(Thread.currentThread().getName() + " 执行工作开始");                futureTask.run();                System.out.println(Thread.currentThread().getName() + " 执行工作完结");            } finally {                endGate.countDown();            }        }    }).start();    // 期待工作    new Thread(new Runnable() {        @Override        public void run() {            try {                submitGate.await();                Thread.sleep(1000);// 期待 futureTask.run() 执行一段时间后再获取后果                System.out.println(Thread.currentThread().getName() + " 获取工作后果开始");                String result = futureTask.get();                System.out.println(Thread.currentThread().getName() + " 获取工作后果完结 " + result);            } catch (Exception e) {                System.out.println(Thread.currentThread().getName() + " 获取工作后果失败 " + e.getMessage());                e.printStackTrace();            } finally {                endGate.countDown();            }        }    }).start();    // 勾销工作    new Thread(new Runnable() {        @Override        public void run() {            try {                submitGate.await();                Thread.sleep(2000);// 期待 futureTask.get() 执行一段时间后再勾销工作                System.out.println(Thread.currentThread().getName() + " 勾销工作开始");                boolean cancel = futureTask.cancel(true);                System.out.println(Thread.currentThread().getName() + " 勾销工作完结 " + cancel);            } catch (Exception e) {                System.out.println(Thread.currentThread().getName() + " 勾销工作失败 " + e.getMessage());                e.printStackTrace();            } finally {                endGate.countDown();            }        }    }).start();    endGate.await();}

执行后果：

Thread-0 执行工作开始Thread-1 获取工作后果开始Thread-2 勾销工作开始Thread-2 勾销工作完结 trueThread-0 执行工作完结Thread-1 获取工作后果失败 nulljava.util.concurrent.CancellationException    at java.util.concurrent.FutureTask.report(FutureTask.java:121)    at java.util.concurrent.FutureTask.get(FutureTask.java:192)    at com.sumkor.pool.FutureTest$6.run(FutureTest.java:129)    at java.lang.Thread.run(Thread.java:745)

阐明：

线程 A 启动工作一段时间后，线程 B 来获取工作后果，进入期待。
随后线程 C 勾销工作，将线程 A 中断（线程 A 不会抛异样，因为 FutureTask#cancel 先一步批改了 state 导致 FutureTask#setException 中 CAS 失败）。
此时线程 B 在期待中被唤醒（由线程 C 唤醒，查看到 state 为 INTERRUPTED）并抛出异样 CancellationException。

7. 总结

FutureTask 实现了 Runnable 和 Future 接口，是一个可勾销的异步工作。
FutureTask 中的工作具备 7 种状态，多个线程之间通过该状态来操作工作，如判断工作是否已实现、勾销工作、获取工作后果。
FutureTask 中只有工作不是 NEW 状态，就示意工作曾经执行结束或者不再执行了，并没有示意“工作正在执行中”的状态。
FutureTask 中应用链表和 CAS 机制构建一个并发平安的栈，用于存储期待获取工作后果的线程。
FutureTask 在期待获取工作后果时，依旧会阻塞主线程，违反了异步的初衷。JDK 8 引入了 CompletableFuture，利用回调机制来做到异步获取工作后果。

作者：Sumkor
链接：https://segmentfault.com/a/11...