Java并發(fā)包FutureTask詳解
我們已經(jīng)知道了所有提交給jdk線程池的任務(wù)都會(huì)被封裝成一個(gè)FutureTask對(duì)象。線程池執(zhí)行的其實(shí)是FutureTask中的run方法。
類圖
image.png
可以看到FutureTask實(shí)現(xiàn)了Future和Runnable兩個(gè)接口，說明它既可以當(dāng)做任務(wù)提交給線程池，也可作為Future查詢?nèi)蝿?wù)執(zhí)行情況或者是取消任務(wù)。
成員變量
/* 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;
這是用于定義任務(wù)狀態(tài)的變量。
- 0-初始
- 1-執(zhí)行中
- 2-已完成
- 3-異常
- 4-已取消
- 5-中斷中
- 6-被中斷
/** 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;
- callable-就是我們傳遞進(jìn)來的原始的任務(wù)
- outcome-任務(wù)執(zhí)行的結(jié)果
- runner-執(zhí)行該任務(wù)的線程
- waiters-等待任務(wù)完成的線程
構(gòu)造函數(shù)
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
FutureTask一共有兩個(gè)構(gòu)造函數(shù)，第一個(gè)構(gòu)造函數(shù)接收一個(gè)Callable對(duì)象，第二個(gè)構(gòu)造函數(shù)接收一個(gè)Runnable對(duì)象和一個(gè)泛型對(duì)象result，這個(gè)構(gòu)造函數(shù)中調(diào)用Executors的callable方法將Runnable對(duì)象包裝成一個(gè)Callable對(duì)象。
Executors的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;
}
}
這是將Runnable轉(zhuǎn)化為Callable對(duì)象的過程，其實(shí)很簡單，由于Runnable并沒有返回值，而Callable需要返回值，因此就直接拿我們傳遞的result作為返回值了。
注意到，構(gòu)造函數(shù)中將任務(wù)的狀態(tài)置為NEW。
成員函數(shù)
public boolean isCancelled() {
return state >= CANCELLED;
}
public boolean isDone() {
return state != NEW;
}
第一個(gè)方法判斷任務(wù)是否已經(jīng)被取消了?？梢钥吹?，5，6狀態(tài)均視為已取消。
第二個(gè)方法判斷任務(wù)是否已經(jīng)完成。只要不是初始狀態(tài)都視為已完成。
public boolean cancel(boolean mayInterruptIfRunning) {
//如果state==NEW，說明任務(wù)還沒開始，此時(shí)只需要根據(jù)傳遞的參數(shù)將其狀態(tài)置為中斷中或者已取消即可
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
//任務(wù)并不是初始狀態(tài)，說明已經(jīng)開始執(zhí)行了
try { // in case call to interrupt throws exception
if (mayInterruptIfRunning) {
try {
//嘗試中斷
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
//將狀態(tài)置為已中斷
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}
該方法用于取消任務(wù)。參數(shù)的意義是如果任務(wù)已經(jīng)開始，是否嘗試中斷。
/**
* Removes and signals all waiting threads, invokes done(), and
* nulls out callable.
*/
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
}
喚醒（unpark方法）所有等待該任務(wù)的線程，并從等待列表中移除。最后調(diào)用了done方法。該方法在FutureTask為空，主要目的是便于子類定制自己的行為。接著將任務(wù)置為null。
再來看看最常用的get方法。
public V get() throws InterruptedException, ExecutionException {
int s = state;
//此時(shí)任務(wù)尚未完成，進(jìn)入到等待中
if (s <= COMPLETING)
s = awaitDone(false, 0L);
//
return report(s);
}
/**
* @throws CancellationException {@inheritDoc}
*/
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);
}
/**
* 該方法只用于獲取已經(jīng)正常完成的任務(wù)的返回值 其他情況都會(huì)拋異常
* Returns result or throws exception for completed task.
*
* @param s completed state value
*/
@SuppressWarnings("unchecked")
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);
}
第一個(gè)get方法會(huì)一直阻塞到任務(wù)完成。第二個(gè)get方法會(huì)至多阻塞指定的時(shí)長。來看看兩個(gè)get中都調(diào)用的一個(gè)方法。
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
//計(jì)算等待的截止時(shí)間點(diǎn)
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
//自旋開始
for (;;) {
//如果當(dāng)前線程被中斷了則移除所有在等待的線程
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
// 如果任務(wù)已經(jīng)結(jié)束了 則返回當(dāng)前狀態(tài)
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
//如果任務(wù)在進(jìn)行中(狀態(tài)馬上就會(huì)變)，則讓出cpu，等待下次cpu時(shí)間片，暫時(shí)不要time out
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
//頭一次自旋則創(chuàng)建一個(gè)等待節(jié)點(diǎn)，繼續(xù)自旋
else if (q == null)
q = new WaitNode();
else if (!queued)
//將當(dāng)前線程加入到等待隊(duì)列中，加入到隊(duì)列的頭部
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
//到這里時(shí) 當(dāng)前線程已經(jīng)加入到等待隊(duì)列中了
else if (timed) {
//如果是設(shè)置了時(shí)間限制，則計(jì)算到時(shí)間了沒，由于前面讓出了CPU時(shí)間片，所以有可能再次執(zhí)行時(shí)已經(jīng)過點(diǎn)了
nanos = deadline - System.nanoTime();
//到點(diǎn)后還沒執(zhí)行完則移除q之后的所有等待線程，并返回當(dāng)前狀態(tài)
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
//掛起當(dāng)前線程
LockSupport.parkNanos(this, nanos);
}
else
//掛起當(dāng)前線程
LockSupport.park(this);
}
}
再來看看核心方法
public void run() {
// 前面半句是判斷任務(wù)的狀態(tài)是不是初始狀態(tài)，只有初始狀態(tài)的任務(wù)才能執(zhí)行run
// 后面半句是判斷執(zhí)行該任務(wù)的線程是否為空，如果不為空則將當(dāng)前線程賦值給runner
// 如果任務(wù)不為初始狀態(tài) 或者 已經(jīng)有指定的執(zhí)行線程了 就直接return，這說明已經(jīng)有線程在執(zhí)行該任務(wù)了
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
//這里就比較簡單了，直接用指定的線程執(zhí)行任務(wù)
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
//異常的情況下
result = null;
ran = false;
setException(ex);
}
//正常完成情況下 調(diào)用set方法
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);
}
}
protected void setException(Throwable t) {
//先將任務(wù)狀態(tài)置為完成中
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//將任務(wù)的返回值置為異常對(duì)象
outcome = t;
//將任務(wù)置為異常狀態(tài)
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
//喚醒所有等待線程
finishCompletion();
}
}
/** 保證所有的強(qiáng)制中斷只下發(fā)到正在執(zhí)行中或者是重置后的任務(wù)中
* Ensures that any interrupt from a possible cancel(true) is only
* delivered to a task while in run or runAndReset.
*/
private void handlePossibleCancellationInterrupt(int s) {
// It is possible for our interrupter to stall before getting a
// chance to interrupt us. Let's spin-wait patiently.
if (s == INTERRUPTING)
//如果任務(wù)狀態(tài)是中斷中，則一直等待
while (state == INTERRUPTING)
Thread.yield(); // wait out pending interrupt
// assert state == INTERRUPTED;
// We want to clear any interrupt we may have received from
// cancel(true). However, it is permissible to use interrupts
// as an independent mechanism for a task to communicate with
// its caller, and there is no way to clear only the
// cancellation interrupt.
//
// Thread.interrupted();
}
FutureTask還提供了一個(gè)可復(fù)用自身的方法。我們注意到run方法中會(huì)判斷任務(wù)是否是初始狀態(tài)，如果不是則不予執(zhí)行。這意味著一個(gè)已經(jīng)完成的任務(wù)是不可能再次被執(zhí)行的。而FutureTask的runAndReset方法則是讓任務(wù)實(shí)現(xiàn)復(fù)用。
protected boolean runAndReset() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return false;
boolean ran = false;
int s = state;
try {
Callable<V> c = callable;
if (c != null && s == NEW) {
try {
c.call(); // don't set result
ran = true;
} catch (Throwable ex) {
setException(ex);
}
}
} 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
s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
return ran && s == NEW;
}
可以看到幾乎和run方法一樣，但是不同點(diǎn)在于任務(wù)執(zhí)行完畢后并沒有調(diào)用set方法，即并沒有改變?nèi)蝿?wù)的狀態(tài)，也沒有將任務(wù)的執(zhí)行結(jié)果保存在outcome中。方法僅在任務(wù)成功執(zhí)行且為初始狀態(tài)時(shí)返回true。
最后是該類中的一個(gè)內(nèi)部類
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
其實(shí)就是一個(gè)單向鏈表，保存等待中的線程。
FutureTask中最復(fù)雜的方法
/**
* Tries to unlink a timed-out or interrupted wait node to avoid
* accumulating garbage. Internal nodes are simply unspliced
* without CAS since it is harmless if they are traversed anyway
* by releasers. To avoid effects of unsplicing from already
* removed nodes, the list is retraversed in case of an apparent
* race. This is slow when there are a lot of nodes, but we don't
* expect lists to be long enough to outweigh higher-overhead
* schemes.
*/
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;
}
}
}
該方法旨在并發(fā)情況下能清除掉所有等待某個(gè)FutureTask完成的線程節(jié)點(diǎn)。
FutureTask全程沒有加鎖，全部是調(diào)用Unsafe類的cas方法來實(shí)現(xiàn)的。
總結(jié)
FutureTask將原始的任務(wù)（Callable或者Runnable）封裝了一層，加入了狀態(tài)位，并且自己維護(hù)狀態(tài)。最終提交到線程池的對(duì)象其實(shí)是一個(gè)FutureTask，由于FutureTask實(shí)現(xiàn)了Runnable接口，因此線程池執(zhí)行的其實(shí)是FutureTask的run方法。任務(wù)的狀態(tài)變更都是FutureTask自己完成的，線程池對(duì)于FutureTask內(nèi)部的狀態(tài)一無所知。一個(gè)FutureTask在執(zhí)行前被取消，并不意味著線程池不再派出線程去執(zhí)行FutureTask，線程池照樣會(huì)派出線程去執(zhí)行該FutureTask，只是在執(zhí)行FutureTask的run方法時(shí)，F(xiàn)utureTask判斷自己已經(jīng)被取消了，就直接return了，不再執(zhí)行包在其中的原始任務(wù)了。
