概念

Callable类的定义

@FunctionalInterface
public interface Callable<V> {V call() throws Exception;
}

Runnable类的定义

@FunctionalInterface
public interface Runnable {public abstract void run();
}

Future类的定义

public interface Future<V> {boolean cancel(boolean mayInterruptIfRunning);boolean isCancelled();boolean isDone();V get() throws InterruptedException, ExecutionException;V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

Callable Runnable Future 都是为异步执行设计的接口类。Callable与Runnable接口的区别是Callable有返回值，并且会抛出异常信息，Runnable没有返回值，也不允许抛出异常。Future则可以判断任务是否执行完，是否取消，以及取消当前任务和获取结果。

使用实例

Runnable实例

Runnable runnable = new Runnable() {@Overridepublic void run() {// do business jobSystem.out.println("thread run = " + Math.random());}
};new Thread(runnable).start();
//  new Thread(runnable).run();System.out.println("done");

定义了一个runnable实例放入Thread并且调用start()方法就可以启动一个线程来执行。这里注意是调用Thread.start()方法，不能调用Thread.run()方法。run()其实是串行执行的，start()才会启动线程异步执行。

start()和run()区别

/* What will be run. */
private Runnable target;public Thread(Runnable target) {init(null, target, "Thread-" + nextThreadNum(), 0);
}@Override
public void run() {if (target != null) {target.run();}
}/*** Causes this thread to begin execution; the Java Virtual Machine* calls the <code>run</code> method of this thread.*/
public synchronized void start() {if (threadStatus != 0)throw new IllegalThreadStateException();group.add(this);boolean started = false;try {start0();started = true;} finally {try {if (!started) {group.threadStartFailed(this);}} catch (Throwable ignore) {}}
}private native void start0();

阅读Thread类的源代码，构造函数将runnable对象赋值给内部的target变量。调用run()就是直接调用target对象的run()。调用start()其实是调用start0()，而start0()是一个native本地方法，由JVM调用操作系统类库来启动线程。

Callable+Future实例

Callable<Double> callable = new Callable<Double>() {@Overridepublic Double call() throws Exception {// do business jobreturn Math.random();}
};FutureTask<Double> future = new FutureTask<>(callable);new Thread(future).start();// do business jobSystem.out.println("future result = " + future.get());
System.out.println("future result = " + future.get(100, TimeUnit.MILLISECONDS));

Callable必须要结合Future来一起使用，声明一个callable实例，通过这个实例再生成一个FutureTask类型的实例放入Thread执行。当调用future.get()的时候，如果future task已经执行完毕则可以获得结果，否则堵塞当前线程直到线程执行完并且返回结果，future.get(long, TimeUnit)支持获取执行结果超时限制。

为什么一定要生成这个FutureTask实例？原因是Thread的构造方法只接受Runnable类型的变量

Thread(Runnable target) {...}
Thread(Runnable target, AccessControlContext acc) {...}
Thread(Runnable target, String name) {...}

再看一下FutureTask的定义

public class FutureTask<V> implements RunnableFuture<V> {...public FutureTask(Callable<V> callable) {if (callable == null)throw new NullPointerException();this.callable = callable;this.state = NEW;}...
}public interface RunnableFuture<V> extends Runnable, Future<V> {void run();
}

FutureTask继承自RunnableFuture，而RunnableFuture同时继承了Runnable和Future。FutureTask是Future的实现类又是Runnable的实现类，可以得出的结论是Future提供的方法都是基于Callable接口实现的。

Future 源码阅读

Future线程状态

Future内部定义了一组线程的运行状态

/*** 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;

从注释里面看出来一共有4种状态的变化

1. NEW（初始化）-> COMPLETING（运行中）-> NORMAL（完成状态）
2. NEW（初始化）-> COMPLETING（运行中）-> EXCEPTIONAL（运行发生错误）
3. NEW（初始化）-> CANCELLED（还未运行已经被取消）
4. NEW（初始化）-> INTERRUPTING（运行中被取消）-> INTERRUPTED（被取消状态）

Future内部变量

/** 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;static final class WaitNode {volatile Thread thread;volatile WaitNode next;WaitNode() { thread = Thread.currentThread(); }
}

callable即实际运行的callable对象，outcome是运行结果存储的变量，waiters是一个链表结构的东西，其实是一个对象拥有下面一个对象的指针，后面会解释（注释上说这个叫Treiber stack）。

Future.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 = null;int s = state;if (s >= INTERRUPTING)handlePossibleCancellationInterrupt(s);}
}protected void set(V v) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome = v;UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final statefinishCompletion();}
}

run()方法实际调用的是callble.call()方法，获取到返回值之后调用set()方法，set()方法通过CAS将线程状态从NEW设置为COMPLETING，再将返回值设置到outcome变量，然后将线程状态设置为NORMAL完成的状态。最后的finishCompletion()方法下面再讲解。

Future.get()

public V get() throws InterruptedException, ExecutionException {int s = state;if (s <= COMPLETING)s = awaitDone(false, 0L);return report(s);
}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);
}

get()方法判断线程状态，如果线程状态不是小于等于COMPLETING的状态调用report()，report()方法判断线程状态为NORMAL就直接返回outcome的值，如果线程状态为CANCELLED就抛出CancellationException异常。

如果线程状态小于等于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);}
}static final class WaitNode {volatile Thread thread;volatile WaitNode next;WaitNode() { thread = Thread.currentThread(); }
}

awaitDone方法内有一个循环，循环内一串判断条件

1. 如果线程状态大于COMPLETING，将q（waitNode）变量的thread设置为null，然后把线程状态返回出去
2. 如果线程状态等于COMPLETING，调用Thread.yield()让出当前线程的CPU使用时间
3. 如果q==null，创建一个新的WaitNode节点
4. 如果queued==false（还未被加入等待队列），使用CAS操作将上一步创建的waitNode设置为waiters链表的表头
5. 如果有超时限制，判断是否超时，如果超时，将waiters链表的节点移除，如果未超时，调用LockSupport.parkNanos()阻塞线程
6. 以上都不满足，调用LockSupport.park()阻塞线程

循环内的判断条件都是排他的，这个循环一般会循环三次。

• 第一次循环执行q==null的条件，创建WaitNode节点。
• 第二次循环执行!queued条件，将刚才创建的waitNode节点设置为waiters链表的表头。WaitNode类存了一个线程的引用以及下一个WaitNode节点的引用，这是一个单向链表的数据结构。
• 第三次循环执行到LockSupport.park*()阻塞线程

为什么需要一个链表？

我能想到的是在多个线程一起调用get()/get(timeout)方法的时候才需要这个链表，因为get()/get(timeout)在task处于非完成状态时是调用LockSupport.park*()阻塞线程的，在多个线程进行get操作，需要一个链表来维护这些线程，一会在task执行完或者出现异常的时候，会在这个链表中找到正在被堵塞的线程调用LockSupport.unpark()来解除堵塞。因为这样的机制才需要一个链表。

再回顾刚才的FutureTask.run()方法，出现异常的时候会调用setException(ex)，线程执行完之后会执行set(result)。

protected void setException(Throwable t) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome = t;UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final statefinishCompletion();}
}protected void set(V v) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome = v;UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final statefinishCompletion();}
}

出现异常的时候将线程的最终状态设置为EXCEPTIONAL，正常结束的时候将线程的最终状态设置为NORMAL，然后都会调用finishCompletion()。

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
}protected void done() { }

这个方法会循环这个waiters链表，取出里面正在等待的线程逐个调用LockSupport.unpark(t)来解除堵塞。通过CAS操作将waiters设置为null。

这里还有一个done()方法，方法体是空的。可以被子类重写，做一些线程执行完成之后的操作。这个也可以会称为回调函数。