JDK源码分析-FutureTask

Posted 2021-04-05 WriteOnRead

概述

FutureTask 是一个可取消的、异步执行任务的类，它的继承结构如下：

它实现了 RunnableFuture 接口，而该接口又继承了 Runnable 接口和 Future 接口，因此 FutureTask 也具有这两个接口所定义的特征。FutureTask 的主要功能：

1. 异步执行任务，并且任务只执行一次；

2. 监控任务是否完成、取消任务；

3. 获取任务执行结果。

下面分析其代码实现。

代码分析

分析 FutureTask 的代码之前，先看下它实现的接口。RunnableFuture 接口定义如下：

public interface RunnableFuture<V> extends Runnable, Future<V> { /** * Sets this Future to the result of its computation * unless it has been cancelled. */ void run();}

RunnableFuture 接口继承了 Runnable 接口和 Future 接口，而 Runnable 接口只有一个 run 方法，这里不再赘述。下面分析 Future 接口。

Future 接口

Future 接口方法定义如下：

主要方法分析：

/* * 尝试取消执行任务。若任务已完成、已取消，或者由于其他某些原因无法取消，则尝试失败。 * 若成功，且调用该方法时任务未启动，则此任务不会再运行； * 若任务已启动，则根据参数 mayInterruptIfRunning 决定是否中断该任务。 */boolean cancel(boolean mayInterruptIfRunning);
// 若该任务正常结束之前被取消，则返回 trueboolean isCancelled();
/* * 若该任务已完成，则返回 true * 这里的“完成”，可能是由于正常终止、异常，或者取消，这些情况都返回 true */boolean isDone();
// 等待计算完成（如果需要），然后获取结果V get() throws InterruptedException, ExecutionException;
// 如果需要，最多等待计算完成的给定时间，然后检索其结果（如果可用）// PS: 该方法与前者的区别在于加了超时等待V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;

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;

其中 state 表示任务的状态，总共有 7 种，它们之间的状态转换可能有以下 4 种情况：

1. 任务执行正常：NEW -> COMPLETING -> NORMAL

2. 任务执行异常：NEW -> COMPLETING -> EXCEPTIONAL

3. 任务取消：NEW -> CANCELLED

4. 任务中断：NEW -> INTERRUPTING -> INTERRUPTED

示意图：

在分析其他成员变量之前，先看一个内部嵌套类 WaitNode：

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

代码比较简单，就是对 Thread 的封装，可以理解为单链表的节点。

其他成员变量：

/** The underlying callable; nulled out after running */// 提交的任务private Callable<V> callable;
/** The result to return or exception to throw from get() */// 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 */// 等待线程的 Treiber 栈private volatile WaitNode waiters;

其中 waiters 是一个 Treiber 栈，简单来说，就是由单链表组成的线程安全的栈，如图所示：

构造器

// 创建一个 FutureTask 对象，在运行时将执行给定的 Callablepublic FutureTask(Callable<V> callable) { if (callable == null) throw new NullPointerException(); this.callable = callable; this.state = NEW; // ensure visibility of callable}
// 创建一个 FutureTask，在运行时执行给定的 Runnable，// 并安排 get 将在成功完成时返回给定的结果public FutureTask(Runnable runnable, V result) { this.callable = Executors.callable(runnable, result); this.state = NEW; // ensure visibility of callable}

这两个构造器分别传入 Callable 对象和 Runnable 对象（适配为 Callable 对象），然后将其状态初始化为 NEW。

run: 执行任务

public void run() {    // 使用 CAS 进行并发控制，防止任务被执行多次 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 {                // 调用 Callable 的 call 方法执行任务 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); }}

set & setException: 更新状态值，唤醒栈中等待的线程

protected void set(V v) { // CAS 将 state 修改为 COMPLETING，该状态是一个中间状态    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { outcome = v; // 输出结果赋值 // 将 state 更新为 NORMAL        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state finishCompletion(); }}
protected void setException(Throwable t) { // CAS 将 state 修改为 COMPLETING，该状态是一个中间状态     if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { outcome = t; // 输出结果赋值 // 将 state 更新为 EXCEPTIONAL         UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state finishCompletion(); }}

这两个方法的操作类似，都是更新 state 的值并给返回结果 outcome 赋值，然后执行结束操作 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; // 唤醒 WaitNode 封装的线程 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}

finishCompletion 方法的作用就是唤醒栈中所有等待的线程，并清空栈。其中的 done 方法实现为空：

protected void done() { }

子类可以重写该方法实现回调功能。

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

这两个方法都是获取任务执行的结果，原理也基本一样，区别在于后者有超时等待（超时会抛出 TimeoutException 异常）。

awaitDone: 等待任务执行完成

// Awaits completion or aborts on interrupt or timeout.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;        // s > COMPLETING 表示任务已执行完成（包括正常执行、异常等状态）        // 则返回对应的状态值 if (s > COMPLETING) { if (q != null) q.thread = null; return s; }        // s == COMPLETING 是一个中间状态，表示任务尚未完成        // 这里让出 CPU 时间片 else if (s == COMPLETING) // cannot time out yet Thread.yield();        // 执行到这里，表示 s == NEW，将当前线程封装为一个 WaitNode 节点 else if (q == null) q = new WaitNode();        // 这里表示 q 并未入栈，CAS 方式将当 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); }}

该方法的主要判断步骤如下：

1. 若线程被中断，则响应中断；

2. 若任务已完成，则返回状态值；

3. 若任务正在执行，则让出 CPU 时间片；

4. 若任务未执行，则将当前线程封装为 WaitNode 节点；

5. 若 WaitNode 未入栈，则执行入栈；

6. 若已入栈，则将线程挂起。

以上步骤是循环执行的，其实该方法的主要作用就是：当任务执行完成时，返回状态值；否则将当前线程挂起。

removeWaiter: 移除栈中的节点

private void removeWaiter(WaitNode node) { if (node != null) { node.thread = null; retry: for (;;) { // restart on removeWaiter race for (WaitNode pred = null, q = waiters, s; q != null; q = s) { s = q.next; if (q.thread != null) pred = q; else if (pred != null) { pred.next = s; if (pred.thread == null) // check for race continue retry; }                else if (!UNSAFE.compareAndSwapObject(this, waitersOffset, q, s)) continue retry; } break; } }}

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

该方法就是对返回结果的包装，无论是正常结束或是抛出异常。

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

场景举例

FutureTask 适合多线程执行一些耗时的操作，然后获取执行结果。下面结合线程池简单分析其用法，示例代码如下（仅供参考）：

public class FutureTaskTest { public static void main(String[] args) throws Exception { ExecutorService executorService = Executors.newFixedThreadPool(5); List<FutureTask<Integer>> taskList = new ArrayList<>(); for (int i = 0; i < 10; i++) { int finalI = i; FutureTask<Integer> futureTask = new FutureTask<>(() -> { // 模拟耗时任务 TimeUnit.SECONDS.sleep(finalI * 2); System.out.println(Thread.currentThread().getName() + " 计算中……"); return finalI * finalI; }); taskList.add(futureTask);            executorService.submit(futureTask); // 提交到线程池 }
        System.out.println("任务全部提交，主线程做其他操作");        // 获取执行结果 for (FutureTask<Integer> futureTask : taskList) { Integer result = futureTask.get(); System.out.println("result-->" + result);        }        // 关闭线程池 executorService.shutdown(); }}

小结

FutureTask 是一个封装任务（Runnable 或 Callable）的类，可以异步执行任务，并获取执行结果，适用于耗时操作场景。

参考链接：

http://www.hchstudio.cn/article/2017/2b8f/

https://segmentfault.com/a/1190000016572591

https://www.jianshu.com/p/43dab9b7c25b