Future、ExecutorService 源码解析

Thread 或者 Runnable 方法开启的线程是没有返回值，如果我们需要子线程计算，取得结果后返回，可以用 Callable。

整体架构

关于线程 API 之间关系的依赖图：

我们首先看一个 demo，这个场景说的是我们往线程池里面提交一个有返回值的线程，代码如下：

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(
        3, 3, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>());
    FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
        @Override
        public String call() throws Exception {
            return "子线程" + Thread.currentThread().getName();
        }
    });
    executor.submit(futureTask);
    String result = futureTask.get();
    System.out.println(result);
}

从这个 demo 中，我们可以看出：

1. Callable 定义我们需要做的事情，可以有返回值
2. FutureTask 我们叫做任务，参数是 Callable，是对 Callable 的包装，方便线程池的使用
3. 最后通过 FutureTask.get() 得到子线程的结果

Callable

Callable 是一个接口，约定了线程要做的事情，和 Runnable 一样，只不过这个线程任务是有返回值的，我们来看下接口定义：

@FunctionalInterface
public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

返回值是一个泛型，可以定义成任何类型，但我们使用的时候，都不会直接使用 Callable，而是会结合 FutureTask 一起使用。

FutureTask

FutureTask 我们可以当做是线程运行的具体任务，从上图中我们可以看到 FutureTask 实现了 RunnableFuture 接口

public class FutureTask<V> implements RunnableFuture<V> {}

而 RunnableFuture 又实现了 Runnable，Future 两个接口

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 两个接口，接下来我们先看 Future，再看 RunnableFuture，最后看 FutureTask。

Future

Callable 是可以返回子线程执行结果的，在获取结果的时候，就需要用到 Future 接口了。

Future 接口注释上写了这些：

1. 定义了异步计算的接口，提供了计算是否完成的 check、等待完成和取回等多种方法；
2. 如果想得到结果可以使用 get 方法，此方法（无参方法）会一直阻塞到异步任务计算完成；
3. 取消可以使用 cancel 方法，但一旦任务计算完成，就无法被取消了。

Future 接口定义了这些方法：

public interface Future<V> {
	// 取消任务
    // 如果任务已经成功了，或者已经取消了，是无法取消的，会直接返回true
    // 如果任务还没开始，发起取消是可以取消成功的
    // 如果任务已经在运行，mayInterruptIfRunning为true的话，可以打断运行中的线程，false不能打断直接返回
    boolean cancel(boolean mayInterruptIfRunning);

    // 判断是否被取消
    boolean isCancelled();

    // 判断是否运行结束（包括被取消）
    boolean isDone();

	// 阻塞获取结果，被取消和中断会抛对应的异常
    V get() throws InterruptedException, ExecutionException;

    // 阻塞获取结果，超时会抛异常
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

RunnableFuture

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

RunnableFuture 接口的最大目的，是让 Future 可以对 Runnable 进行管理，可以取消 Runnable，查看 Runnable 是否完成等等。

统一 Callable 和 Runnable

我们现在清楚了，新建任务有两种方式，一种是无返回值的 Runnable，一种是有返回值的 Callable，但对 Java 其他 API 来说使用起来并不是很方便，没有一个统一的接口，比如说线程池在提交任务时，是不是应该针对 Runnable 和 Callable 两种情况提供不同的实现思路呢？所以 FutureTask 出现了，FutureTask 实现了 RunnableFuture 接口，又集合了 Callable（Callable 是 FutureTask 的属性），还提供了两者一系列的转化方法，这样 FutureTask 就统一了 Callable 和 Runnable，我们一起来细看下。

FutureTask 的类定义

public class FutureTask<V> implements RunnableFuture<V> {}

从类定义上可以看出来 FutureTask 实现了 RunnableFuture 接口，也就是说间接实现了 Runnnable 接口（RunnableFuture 实现了 Runnnable 接口），就是说 FutureTask 本身就是个 Runnnable，同时 FutureTask 也实现了 Future，也就是说 FutureTask 具备对任务进行管理的功能（Future 具备对任务进行管理的功能）。

FutureTask 的属性

我们一起来看下 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;//任务被打断成功
 
// 组合了 Callable 
private Callable<V> callable;
// 异步线程返回的结果
private Object outcome; 
// 当前任务所运行的线程
private volatile Thread runner;
// 记录调用 get 方法时被等待的线程
private volatile WaitNode waiters;

从属性上我们明显看到 Callable 是作为 FutureTask 的属性之一，这也就让 FutureTask 具备了转化 Callable 和 Runnable 的功能，接着我们看下 FutureTask 的构造器，看看两者是如何转化的。

FutureTask 的构造器

FutureTask 有两个构造器，分别接受 Callable 和 Runnable，如下：

// 使用 Callable 进行初始化
public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    // 任务状态初始化
    this.state = NEW;       // ensure visibility of callable
}
 
// 使用 Runnable 初始化，并传入 result 作为返回结果。
// Runnable 是没有返回值的，所以 result 一般没有用，置为 null 就好了
public FutureTask(Runnable runnable, V result) {
    // Executors.callable 方法把 runnable 适配成 RunnableAdapter，RunnableAdapter 实现了 callable，所以也就是把 runnable 直接适配成了 callable。
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;       // ensure visibility of callable
}

Runnable 的两个构造器，只有一个目的，就是把入参都转化成 Callable，那么为什么不都转化成 Runnnable 呢？主要是因为 Callable 的功能比 Runnnable 丰富，Callable 有返回值，而 Runnnable 没有。

我们注意到入参是 Runnable 的构造器，会使用 Executors.callable 方法来把 Runnnable 转化成 Callable，Runnnable 和 Callable 两者都是接口，两者之间是无法进行转化的，所以 Java 新建了一个转化类：RunnableAdapter 来进行转化，我们来看下转化的逻辑：

// 转化 Runnable 成 Callable 的工具类
static final class RunnableAdapter<T> implements Callable<T> {
    final Runnable task;
    final T result;
    RunnableAdapter(Runnable task, T result) {
        this.task = task;
        this.result = result;
    }
    public T call() {
        task.run();
        return result;
    }
}

我们可以看到：

1. 首先 RunnableAdapter 实现了 Callable，所以 RunnableAdapter 就是 Callable；
2. 其次 Runnable 是 RunnableAdapter 的一个属性，在构造 RunnableAdapter 的时候会传进来，并且在 call 方法里面调用 Runnable 的 run 方法。

这是一个典型的适配模型，我们要把 Runnable 适配成 Callable，首先要实现 Callable 的接口，接着在 Callable 的 call 方法里面调用被适配对象（Runnable）的方法。

FutureTask 构造器设计很巧妙，将 Runnable 和 Callable 灵活的打通，向内和向外只提供功能更加丰富的 Callable 接口，值得我们学习。

FutureTask 对 Future 接口方法的实现

我们主要看几个关键的方法实现源码。

get

get 有无限阻塞和带超时时间两种方法，我们通常建议使用带超时时间的方法，源码如下：

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);
}
// 等待任务执行完成
private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    // 计算等待终止时间，如果一直等待的话，终止时间为 0
    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;
        }
        // 如果正在执行，当前线程让出 cpu，重新竞争，防止 cpu 飙高
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
            // 如果第一次运行，新建 waitNode，当前线程就是 waitNode 的属性
        else if (q == null)
            q = new WaitNode();
            // 默认第一次都会执行这里，执行成功之后，queued 就为 true，就不会再执行了
            // 把当前 waitNode 当做 waiters 链表的第一个
        else if (!queued)
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                 q.next = waiters, q);
            // 如果设置了超时时间，并过了超时时间的话，从 waiters 链表中删除当前 wait
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                removeWaiter(q);
                return state;
            }
            // 没有过超时时间，线程进入 TIMED_WAITING 状态
            LockSupport.parkNanos(this, nanos);
        }
        // 没有设置超时时间，进入 WAITING 状态
        else
            LockSupport.park(this);
    }
}

get 方法虽然名字叫做 get，但却做了很多 wait 的事情，当发现任务还在进行中，没有完成时，就会阻塞当前进程，等待任务完成后再返回结果值。阻塞底层使用的是 LockSupport.park 方法，使当前线程进入 WAITING 或 TIMED_WAITING 状态。

run

/**
 * run 方法可以直接被调用
 * 也可以开启新的线程进行调用
 */
public void run() {
    // 状态不是任务创建，或者当前任务已经有线程在执行了，直接返回
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        Callable<V> c = callable;
        // 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);
            }
            // 给 outcome 赋值
            if (ran)
                set(result);
        }
    } finally {
        runner = null;
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

run 方法我们再说明几点：

1. run 方法是没有返回值的，通过给 outcome 属性赋值（set(result)），get 时就能从 outcome 属性中拿到返回值；
2. FutureTask 两种构造器，最终都转化成了 Callable，所以在 run 方法执行的时候，只需要执行 Callable 的 call 方法即可，在执行 c.call() 代码时，如果入参是 Runnable 的话， 调用路径为 c.call() -> RunnableAdapter.call() -> Runnable.run()，如果入参是 Callable 的话，直接调用。

cancel

// 取消任务，如果正在运行，尝试去打断
public boolean cancel(boolean mayInterruptIfRunning) {
    if (!(state == NEW &&//任务状态不是创建 并且不能把 new 状态置为取消，直接返回 false
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    // 进行取消操作，打断可能会抛出异常，选择 try finally 的结构
    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;
}

总结

大家现在可以回头看看一开始我们贴出来的图，看看自己照着图能否想起来各个 API 的作用，比如 Callable 是干啥的，FutureTask 又有什么作用，Runnable 和 Calllable 之间又是如何关联起来，几个 API 之间的关系的确很复杂，FutureTask 是关键，通过 FutureTask 把 Runnnable、Callable、Future 都串起来了，使 FutureTask 具有三者的功能，统一了 Runnnable 和 Callable，更方便使用。