引言
上一篇文章，我们认识了Callable、Future和ExecutorService，它们分别代表了任务、任务的结果和任务的执行。遗留的一个问题是ExecutorService的submit方法应该是怎样的逻辑。这篇文章，我们来看这个逻辑的实现。
从submit方法看FutureTask
AbstractExecutorService是ExecutorService的实现类。它给出了ExecutorService中的几个方法例如submit、invokeAll、invokeAny等的实现。我们首先来看参数为Callable的submit方法的实现：
public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask;}
首先，它调用了newTaskFor方法创建了一个RunnableFuture对象，RunnableFuture从名字上看应该和Runnable和Future都有关系。newTaskFor方法如下：
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {return new FutureTask<T>(callable);}
注意，该方法的返回值是RunnableFuture，实际上返回的是FutureTask。可以猜测FutureTask是RunnableFuture的实现类。RunnableFuture的定义如下：
public interface RunnableFuture<V> extends Runnable, Future<V> {void run();}
果然是实现了Runnable和Future接口，也就是RunnableFuture将任务和任务的结果结合了起来。我们再看FutureTask，这个是我们重点关注的类：
public class FutureTask<V> implements RunnableFuture<V> {}
果然是实现了RunnableFuture接口，上面的newTaskFor方法调用的是FutureTask参数是Callable的构造方法，我们找到这个构造方法：
public FutureTask(Callable<V> callable) {if (callable == null)throw new NullPointerException();this.callable = callable;this.state = NEW; // ensure visibility of callable}
它设置了callable和state两个字段，这里，我们看下FutureTask比较重要的字段：
FutureTask重要字段
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是当前任务的状态，有以下几种状态：
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状态，上面我们看到的构造方法，就是将状态设置为NEW。COMPLETING是执行中状态。NORMAL是成功执行完成状态，还有取消、中断等状态。注意，COMPLETING之后的状态都是完成状态。
callable实际上代表的是FutureTask要执行的任务。你可能会有疑问，FutureTask本身就是Runnable，有自己的run方法，为什么还要有Callable呢？这个我们稍后解释。现在，你只要知道FutureTask除了本身实现了Future和Runnable，内部还有一个Callable就可以了。
outcome代表任务执行的返回结果。
runner代表执行这个任务的线程。
WaitNode是一个内部类：
static final class WaitNode {volatile Thread thread;volatile WaitNode next;WaitNode() { thread = Thread.currentThread(); }}
代表等待结果的线程。
为啥会有等待的线程，注意这里的线程不是要执行任务的线程，而是调用FutureTask的get方法获取任务结果的线程，前面我们知道，get方法当任务没有执行完成时，会处于等待状态。并且等待的线程可能不止一个，所以WaitNode有一个next字段，同时每个WaitNode有一个对应的线程。
这几个就是FutureTask的重要字段，下面，我们看下它的几个重要方法。
FutureTask关键方法逻辑
FutureTask就是实现了Runnable和Future，所以它的几个方法也就是Runnable和Future接口的那些方法。我们首先来看Runnable的run方法。
run方法
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 interruptsint s = state;if (s >= INTERRUPTING)handlePossibleCancellationInterrupt(s);}}
首先，它会判断任务的状态是否是NEW，然后原子性地将runner设置为当前线程，也就是执行任务的线程。然后，它会调用内部Callable的call方法来执行任务，所以说FutureTask虽然实现了Runnable接口，但是它的run方法里面实际上执行的是内部的Callable，也就是说callable才是真正要执行的任务。
执行成功之后，它会调用set方法来设置任务的返回结果：
protected void set(V v) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome = v;UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final statefinishCompletion();}}
set方法会将state设置为NORMAL，表示任务正常执行完成，并且将outcome设置为callable返回值，也就是设置任务的返回结果。然后调用了finishCompletion()方法，该方法用来唤醒执行get方法而等待的线程：
private void finishCompletion() {// assert state > COMPLETING;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 gcq = next;}break;}}done();callable = null; // to reduce footprint}
在这个方法中，通过for(;;)不断地找到下一个代表等待线程的WaitNode，然后通过LockSupport.unpark方法唤醒这些线程。我们猜测一下，FutureTask的get方法应该会调用LockSupport.park方法来挂起线程。
get方法
get方法的逻辑如下：
public V get() throws InterruptedException, ExecutionException {int s = state;if (s <= COMPLETING)s = awaitDone(false, 0L);return report(s);}
首先，它判断state是否是未完成状态（NEW和COMPLETING），如果是未完成状态，就需要当前线程等待任务完成，它调用的是awaitDone方法：
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 yetThread.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);}elseLockSupport.park(this);}}
我们这里只看正常的挂起流程，它会构造一个WaitNode节点，然后调用
UNSAFE.compareAndSwapObject(this, waitersOffset,q.next = waiters, q);
将当前节点加到最前面。
然后调用LockSupport.park方法将当前线程挂起。线程的唤醒我们在run方法中已经讲到了。
继续看submit
理解了FuthreTask的主要方法之后，我们继续看AbstractExecutorService类的submit方法：
public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask;}
它只是创建了一个FutureTask，然后调用execute方法并将FutureTask作为参数，execute方法是Executor接口提供的方法，它的参数是Runnble而FutureTask实现了Runnable接口，我们猜测，execute方法里面就会调用FutureTask的run方法（可能是当前线程调用也可能是其他线程调用），execute方法的具体实现依赖于子类。我们在讲述ThreadPoolExecutor时，会重点讲述它的execute方法的实现。
到此为止，我们应该明白了submit方法的大致思路：它用到了FutureTask，这是一个Runnable和Future的结合体，并且内部还有Callable代表要执行的任务。它的run方法会执行Callable的call方法来执行任务，由于涉及到get方法对线程的挂起，run方法在任务执行完成后还会唤醒挂起的线程。FutureTask还给出了Future接口提供的几个方法的实现，get方法会通过LockSupport来挂起线程来等待任务完成。
invokeAll方法
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)throws InterruptedException {if (tasks == null)throw new NullPointerException();ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());boolean done = false;try {for (Callable<T> t : tasks) {RunnableFuture<T> f = newTaskFor(t);futures.add(f);execute(f);}for (int i = 0, size = futures.size(); i < size; i++) {Future<T> f = futures.get(i);if (!f.isDone()) {try {f.get();} catch (CancellationException ignore) {} catch (ExecutionException ignore) {}}}done = true;return futures;} finally {if (!done)for (int i = 0, size = futures.size(); i < size; i++)futures.get(i).cancel(true);}}
这个方法应该不难理解，就是对每个Callable，创建一个FutureTask，然后执行这些FutureTask。注意它还会调用每个FutureTask的get方法来等待任务完成。
小结
FutureTask实现了Callable任务到Future任务结果的映射，它的关键是run方法中对callable的call方法的调用。另外它还给出了Future接口几个方法的实现，get方法在任务没有完成的情况下会将线程挂起，run方法在任务执行完成后会将挂起的线程唤醒。AbstractExecutorService通过FutureTask实现了ExecutorService的几个方法，例如submit、invokeAll等。有了这些知识储备，我们就能更轻松地去理解线程池了。下一篇文章，我们重点理解线程池ThreadPoolExecutor。
若有收获，就点个赞吧
0 人点赞
