一、理解
future模式是多線程開發(fā)中非常常見的一種設(shè)計(jì)模式，它的核心思想是異步調(diào)用。當(dāng)我們需要調(diào)用一個函數(shù)方法時，如果這個函數(shù)執(zhí)行很慢，那么我們就要進(jìn)行等待。但有時候我們可能并不著急著要結(jié)果。因此，我們可以讓被調(diào)用者立即返回，讓他在后臺慢慢處理這個請求。對于調(diào)用者來說，則可以先處理一些其他任務(wù)，在真正需要數(shù)據(jù)的場合再去嘗試獲得需要的數(shù)據(jù)。
對于future模式來說，雖然它無法立即給你需要的數(shù)據(jù)。但是，它會返回給你一個契約，將來，你可以憑借這個契約去重新獲取你需要的信息。
二、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;
}
Future保存異步計(jì)算的結(jié)果,該接口提供以下的方法
1.cancel(boolean mayInterruptIfRunning)： 試圖取消執(zhí)行的任務(wù)，參數(shù)為true時直接中斷正在執(zhí)行的任務(wù)，否則直到當(dāng)前任務(wù)執(zhí)行完成，成功取消后返回true，否則返回false
2.isCancelled：判斷任務(wù)是否在正常執(zhí)行完前被取消的，如果是則返回true
3.isDone: 判斷任務(wù)是否已完成
- get(): 等待計(jì)算結(jié)果的返回(阻塞方法)，如果計(jì)算被取消了則拋出異常
- get(long timeout, TimeUnit unit)：設(shè)定計(jì)算結(jié)果的返回時間，如果在規(guī)定時間內(nèi)沒有返回計(jì)算結(jié)果則拋出異常
三、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()為抽象方法。
四、FutureTask
FutureTask實(shí)現(xiàn)了RunnableFuture接口，即FutureTask對象既是一個Future接口的實(shí)例又是一個Runnable接口的實(shí)例。
private volatile int state;
//表示這是一個新的任務(wù)，或者還沒有執(zhí)行完的任務(wù)，是初始狀態(tài)。
private static final int NEW = 0;
//表示任務(wù)執(zhí)行結(jié)束（正常執(zhí)行結(jié)束，或者發(fā)生異常結(jié)束），但是還沒有將結(jié)果保存到outcome中。是一個中間狀態(tài)。
private static final int COMPLETING = 1;
//示任務(wù)正常執(zhí)行結(jié)束，并且已經(jīng)把執(zhí)行結(jié)果保存到outcome字段中。是一個最終狀態(tài)。
private static final int NORMAL = 2;
//表示任務(wù)發(fā)生異常結(jié)束，異常信息已經(jīng)保存到outcome中，這是一個最終狀態(tài)。
private static final int EXCEPTIONAL = 3;
//任務(wù)在新建之后，執(zhí)行結(jié)束之前被取消了，但是不要求中斷正在執(zhí)行的線程，也就是調(diào)用了cancel(false)，任務(wù)就是CANCELLED狀態(tài)
private static final int CANCELLED = 4;
//任務(wù)在新建之后，執(zhí)行結(jié)束之前被取消了，并要求中斷線程的執(zhí)行，也就是調(diào)用了cancel(true)，這時任務(wù)狀態(tài)就是INTERRUPTING
private static final int INTERRUPTING = 5;
//調(diào)用cancel(true)取消異步任務(wù)，會調(diào)用interrupt()中斷線程的執(zhí)行，然后狀態(tài)會從INTERRUPTING變到INTERRUPTED。
private static final int INTERRUPTED = 6;
狀態(tài)變化：
* Possible state transitions:
* NEW -> COMPLETING -> NORMAL
* NEW -> COMPLETING -> EXCEPTIONAL
* NEW -> CANCELLED
* NEW -> INTERRUPTING -> INTERRUPTED
構(gòu)造器：
//傳入一個Callable對象，可以通過get方法獲取到結(jié)果
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
//傳入一個Runnable對象和一個返回的結(jié)果，該Runnable對象會通過Executors.callable(runnable, result)方法封裝成Callable對象；result作為get方法返回的結(jié)果
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
主要源代碼說明：
// 用來保存計(jì)算任務(wù)的返回結(jié)果，或者執(zhí)行過程中拋出的異常。
/** The result to return or exception to throw from get() */
private Object outcome;
//運(yùn)行傳入的callable對象的線程
private volatile Thread runner;
//WaitNode是FutureTask的內(nèi)部類，表示一個阻塞隊(duì)列，如果任務(wù)還沒有執(zhí)行結(jié)束，那么調(diào)用get()獲取結(jié)果的線程會阻塞，在這個阻塞隊(duì)列中排隊(duì)等待。
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(); }
}
run()方法：
public void run() {
//state != NEW：state狀態(tài)值是否是NEW，不是NEW，說明任務(wù)已經(jīng)被其他線程執(zhí)行，甚至執(zhí)行結(jié)束，或者被取消了，直接返回
//調(diào)用CAS方法，判斷runnerOffset為null的話，就將當(dāng)前線程保存到runnerOffset中(即把當(dāng)前的線程賦值給runner)，
//如果設(shè)置runnerOffset失敗，就直接返回
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 {
// 執(zhí)行Callable任務(wù)，結(jié)果保存到result中
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
//如果執(zhí)行任務(wù)過程中發(fā)生異常，將調(diào)用setException()設(shè)置異常
setException(ex);
}
//任務(wù)正常執(zhí)行結(jié)束調(diào)用set(result)保存結(jié)果
if (ran)
set(result);
}
} finally {
//任務(wù)執(zhí)行結(jié)束，runner設(shè)置為null，表示當(dāng)前沒有線程在執(zhí)行這個任務(wù)了
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
//讀取狀態(tài)，判斷是否在執(zhí)行的過程中，被中斷了，如果被中斷，處理中斷
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
設(shè)置處理異常:
protected void setException(Throwable t) {
//通過CAS操作將當(dāng)前線程的stateOffset(即state)從NEW置為COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//將異常信息保存到outcome
outcome = t;
//設(shè)置狀態(tài)為EXCEPTIONAL
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
設(shè)置處理結(jié)果：
protected void set(V v) {
//把state狀態(tài)從NEW設(shè)置為COMPLETING（0->1）
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//把處理結(jié)果存放在outcome里
outcome = v;
//將當(dāng)前任務(wù)的狀態(tài)改成NORMAL-2
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
get方法：
public V get() throws InterruptedException, ExecutionException {
int s = state;
//狀態(tài)值如果小于COMPLETING，表示未執(zhí)行完成，阻塞線程進(jìn)行等待
//如果任務(wù)還未執(zhí)行完成(任務(wù)狀態(tài)為NEW,從run方法中可以看到只有任務(wù)執(zhí)行結(jié)束，或者發(fā)生異常的時候，state才會被設(shè)置成COMPLETING)
if (s <= COMPLETING)
//調(diào)用awaitDone(false, 0L)，進(jìn)入阻塞狀態(tài)
s = awaitDone(false, 0L);
//返回結(jié)果
return report(s);
}
一般情況下，執(zhí)行任務(wù)的線程和獲取結(jié)果的線程不會是同一個，當(dāng)我們在主線程或者其他線程中，獲取計(jì)算任務(wù)的結(jié)果時，就會調(diào)用get方法，如果這時計(jì)算任務(wù)還沒有執(zhí)行完成，調(diào)用get()的線程就會阻塞等待。
/**
* 等待任務(wù)完成或者中止或者超時
*
* @param timed 是否啟用等待時長
* @param nanos 等待時間
* @return 任務(wù)完成后的狀態(tài)
*/
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
//自驅(qū)
for (;;) {
//如果調(diào)用線程被阻斷了就從等待的線程棧中移除這個等待節(jié)點(diǎn)，然后拋出中斷異常
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
//獲取當(dāng)前任務(wù)的狀態(tài)
int s = state;
//如果當(dāng)前任務(wù)已經(jīng)執(zhí)行完成(正常或異常結(jié)束),不在阻塞,直接返回任務(wù)狀態(tài)
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
//如果任務(wù)結(jié)束,但是最終結(jié)果還沒保存下來,可以暫時讓出CPU
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
//如果等待節(jié)點(diǎn)q=null,就創(chuàng)建一個等待節(jié)點(diǎn)
else if (q == null)
q = new WaitNode();
//如果這個等待節(jié)點(diǎn)還沒有加入等待隊(duì)列，就加入隊(duì)列頭
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
//如果設(shè)置了超時等待時間
else if (timed) {
nanos = deadline - System.nanoTime();
//如果超出了等待時間,停止阻塞,返回當(dāng)前任務(wù)的狀態(tài)
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
//阻塞特定時間
LockSupport.parkNanos(this, nanos);
}
else
//阻塞
LockSupport.park(this);
}
}
private V report(int s) throws ExecutionException {
//獲取當(dāng)前任務(wù)的結(jié)果
Object x = outcome;
//如果任務(wù)的狀態(tài)是正常完成，則返回結(jié)果
if (s == NORMAL)
return (V)x;
//如果任務(wù)被取消了，則拋出取消異常
if (s >= CANCELLED)
throw new CancellationException();
//否則異常結(jié)束，outcome里面保存的是異常結(jié)果,將異常拋出
throw new ExecutionException((Throwable)x);
}
狀態(tài)值不為初始狀態(tài)，表示完成
public boolean isDone() {
return state != NEW;
}
狀態(tài)值為CANCELLED、INTERRUPTING、INTERRUPTED表示已經(jīng)取消，反之為未取消
public boolean isCancelled() {
return state >= CANCELLED;
}
cancel方法：
public boolean cancel(boolean mayInterruptIfRunning) {
//判斷當(dāng)前任務(wù)的狀態(tài)是否為新建狀態(tài)，如果不是說明任務(wù)已經(jīng)正常或異常結(jié)束了，直接返回取消失敗
//如果任務(wù)是NEW狀態(tài)，根據(jù)mayInterruptIfRunning嘗試將任務(wù)狀態(tài)改成CANCELLED或者INTERRUPTING,更改失敗返回false
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try { // in case call to interrupt throws exception
//如果取消的時候嘗試打斷任務(wù)的執(zhí)行
if (mayInterruptIfRunning) {
try {
//獲取執(zhí)行當(dāng)前任務(wù)的線程
Thread t = runner;
if (t != null)
//調(diào)用interrupt方法打斷線程
t.interrupt();
} finally { // final state
//將任務(wù)的狀態(tài)改為INTERRUPTED
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}
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 gc
q = next;
}
break;
}
}
done();
callable = null; // to reduce footprint
}
這個方法在run方法和cancel方法中都有調(diào)用，分析代碼可以知道這個方法是在任務(wù)的狀態(tài)變成終態(tài)的時候會被調(diào)用。該方法主要做了三件事：
1.遍歷waiters等待隊(duì)列，調(diào)用LockSupport.unpark(t)喚醒等待返回結(jié)果的線程，釋放資源。
2.調(diào)用done()，這個方法什么都沒有做，不過子類可以實(shí)現(xiàn)這個方法，做一些額外的操作。
3.設(shè)置callable為null，callable是FutureTask封裝的任務(wù)，任務(wù)執(zhí)行完，釋放資源。
