futuretask源碼分析(推薦)
FutureTask只實(shí)現(xiàn)RunnableFuture接口:
該接口繼承了java.lang.Runnable和Future接口,也就是繼承了這兩個(gè)接口的特性。
1.可以不必直接繼承Thread來生成子類,只要實(shí)現(xiàn)run方法,且把實(shí)例傳入到Thread構(gòu)造函數(shù),Thread就可以執(zhí)行該實(shí)例的run方法了( Thread(Runnable) )。
2.可以讓任務(wù)獨(dú)立執(zhí)行,get獲取任務(wù)執(zhí)行結(jié)果時(shí),可以阻塞直至執(zhí)行結(jié)果完成。也可以中斷執(zhí)行,判斷執(zhí)行狀態(tài)等。
FutureTask是一個(gè)支持取消行為的異步任務(wù)執(zhí)行器。該類實(shí)現(xiàn)了Future接口的方法。
如: 1. 取消任務(wù)執(zhí)行
2. 查詢?nèi)蝿?wù)是否執(zhí)行完成
3. 獲取任務(wù)執(zhí)行結(jié)果(”get“任務(wù)必須得執(zhí)行完成才能獲取結(jié)果,否則會(huì)阻塞直至任務(wù)完成)。
注意:一旦任務(wù)執(zhí)行完成,則不能執(zhí)行取消任務(wù)或者重新啟動(dòng)任務(wù)。(除非一開始就使用runAndReset模式運(yùn)行任務(wù))
FutureTask支持執(zhí)行兩種任務(wù), Callable 或者 Runnable的實(shí)現(xiàn)類。且可把FutureTask實(shí)例交由Executor執(zhí)行。
源碼部分(很簡(jiǎn)單):
public class FutureTask<V> implements RunnableFuture<V> {
/*
* Revision notes: This differs from previous versions of this
* class that relied on AbstractQueuedSynchronizer, mainly to
* avoid surprising users about retaining interrupt status during
* cancellation races. Sync control in the current design relies
* on a "state" field updated via CAS to track completion, along
* with a simple Treiber stack to hold waiting threads.
*
* Style note: As usual, we bypass overhead of using
* AtomicXFieldUpdaters and instead directly use Unsafe intrinsics.
*/
/**
* 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;
/** The underlying callable; nulled out after running */
private Callable<V> callable;
/** 用來存儲(chǔ)任務(wù)執(zhí)行結(jié)果或者異常對(duì)象,根據(jù)任務(wù)state在get時(shí)候選擇返回執(zhí)行結(jié)果還是拋出異常 */
private Object outcome; // non-volatile, protected by state reads/writes
/** 當(dāng)前運(yùn)行Run方法的線程 */
private volatile Thread runner;
/** Treiber stack of waiting threads */
private volatile WaitNode waiters;
/**
* 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);
}
/**
* Creates a {@code FutureTask} that will, upon running, execute the
* given {@code Callable}.
*
* @param callable the callable task
* @throws NullPointerException if the callable is null
*/
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
/**
* Creates a {@code FutureTask} that will, upon running, execute the
* given {@code Runnable}, and arrange that {@code get} will return the
* given result on successful completion.
*
* @param runnable the runnable task
* @param result the result to return on successful completion. If
* you don't need a particular result, consider using
* constructions of the form:
* {@code Future<?> f = new FutureTask<Void>(runnable, null)}
* @throws NullPointerException if the runnable is null
*/
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
//判斷任務(wù)是否已取消(異常中斷、取消等)
public boolean isCancelled() {
return state >= CANCELLED;
}
/**
判斷任務(wù)是否已結(jié)束(取消、異常、完成、NORMAL都等于結(jié)束)
**
public boolean isDone() {
return state != NEW;
}
/**
mayInterruptIfRunning用來決定任務(wù)的狀態(tài)。
true : 任務(wù)狀態(tài)= INTERRUPTING = 5。如果任務(wù)已經(jīng)運(yùn)行,則強(qiáng)行中斷。如果任務(wù)未運(yùn)行,那么則不會(huì)再運(yùn)行
false:CANCELLED = 4。如果任務(wù)已經(jīng)運(yùn)行,則允許運(yùn)行完成(但不能通過get獲取結(jié)果)。如果任務(wù)未運(yùn)行,那么則不會(huì)再運(yùn)行
**/
public boolean cancel(boolean mayInterruptIfRunning) {
if (state != NEW)
return false;
if (mayInterruptIfRunning) {
if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
return false;
Thread t = runner;
if (t != null)
t.interrupt();
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
}
else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
return false;
finishCompletion();
return true;
}
/**
* @throws CancellationException {@inheritDoc}
*/
public V get() throws InterruptedException, ExecutionException {
int s = state;
//如果任務(wù)未徹底完成,那么則阻塞直至任務(wù)完成后喚醒該線程
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);
}
/**
* Protected method invoked when this task transitions to state
* {@code isDone} (whether normally or via cancellation). The
* default implementation does nothing. Subclasses may override
* this method to invoke completion callbacks or perform
* bookkeeping. Note that you can query status inside the
* implementation of this method to determine whether this task
* has been cancelled.
*/
protected void done() { }
/**
該方法在FutureTask里只有run方法在任務(wù)完成后調(diào)用。
主要保存任務(wù)執(zhí)行結(jié)果到成員變量outcome 中,和切換任務(wù)執(zhí)行狀態(tài)。
由該方法可以得知:
COMPLETING : 任務(wù)已執(zhí)行完成(也可能是異常完成),但還未設(shè)置結(jié)果到成員變量outcome中,也意味著還不能get
NORMAL : 任務(wù)徹底執(zhí)行完成
**/
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
/**
* Causes this future to report an {@link ExecutionException}
* with the given throwable as its cause, unless this future has
* already been set or has been cancelled.
*
* <p>This method is invoked internally by the {@link #run} method
* upon failure of the computation.
*
* @param t the cause of failure
*/
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
/**
由于實(shí)現(xiàn)了Runnable接口的緣故,該方法可由執(zhí)行線程所調(diào)用。
**/
public void run() {
//只有當(dāng)任務(wù)狀態(tài)=new時(shí)才被運(yùn)行繼續(xù)執(zhí)行
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 {
//調(diào)用Callable的Call方法
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 interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
/**
如果該任務(wù)在執(zhí)行過程中不被取消或者異常結(jié)束,那么該方法不記錄任務(wù)的執(zhí)行結(jié)果,且不修改任務(wù)執(zhí)行狀態(tài)。
所以該方法可以重復(fù)執(zhí)行N次。不過不能直接調(diào)用,因?yàn)槭莗rotected權(quán)限。
**/
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;
}
/**
* 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)
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();
}
/**
* 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(); }
}
/**
該方法在任務(wù)完成(包括異常完成、取消)后調(diào)用。刪除所有正在get獲取等待的節(jié)點(diǎn)且喚醒節(jié)點(diǎn)的線程。和調(diào)用done方法和置空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
}
/**
阻塞等待任務(wù)執(zhí)行完成(中斷、正常完成、超時(shí))
**/
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 else的順序也是有講究的。
1.先判斷線程是否中斷,中斷則從隊(duì)列中移除(也可能該線程不存在于隊(duì)列中)
2.判斷當(dāng)前任務(wù)是否執(zhí)行完成,執(zhí)行完成則不再阻塞,直接返回。
3.如果任務(wù)狀態(tài)=COMPLETING,證明該任務(wù)處于已執(zhí)行完成,正在切換任務(wù)執(zhí)行狀態(tài),CPU讓出片刻即可
4.q==null,則證明還未創(chuàng)建節(jié)點(diǎn),則創(chuàng)建節(jié)點(diǎn)
5.q節(jié)點(diǎn)入隊(duì)
6和7.阻塞
**/
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 yet
Thread.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);
}
else
LockSupport.park(this);
}
}
/**
* 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;
}
}
}
// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long stateOffset;
private static final long runnerOffset;
private static final long waitersOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = FutureTask.class;
stateOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("state"));
runnerOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("runner"));
waitersOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("waiters"));
} catch (Exception e) {
throw new Error(e);
}
}
}
總結(jié)
以上就是本文關(guān)于futuretask源碼分析(推薦)的全部?jī)?nèi)容,希望對(duì)大家有所幫助。感興趣的朋友可以參閱:Java利用future及時(shí)獲取多線程運(yùn)行結(jié)果、淺談Java多線程處理中Future的妙用(附源碼)、futuretask用法及使用場(chǎng)景介紹等,有什么問題可以隨時(shí)留言,歡迎大家一起交流討論。
相關(guān)文章
SpringBoot整合Xxl-job實(shí)現(xiàn)定時(shí)任務(wù)的全過程
XXL-JOB是一個(gè)分布式任務(wù)調(diào)度平臺(tái),其核心設(shè)計(jì)目標(biāo)是開發(fā)迅速、學(xué)習(xí)簡(jiǎn)單、輕量級(jí)、易擴(kuò)展,下面這篇文章主要給大家介紹了關(guān)于SpringBoot整合Xxl-job實(shí)現(xiàn)定時(shí)任務(wù)的相關(guān)資料,需要的朋友可以參考下2022-01-01
java中樂觀鎖與悲觀鎖區(qū)別及使用場(chǎng)景分析
本文主要介紹了java中樂觀鎖與悲觀鎖區(qū)別及使用場(chǎng)景分析,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2024-08-08
SpringBoot集成Tomcat服務(wù)架構(gòu)配置
這篇文章主要為大家介紹了SpringBoot集成Tomcat服務(wù)架構(gòu)配置,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-02-02
SpringCloud @FeignClient參數(shù)的用法解析
這篇文章主要介紹了SpringCloud @FeignClient參數(shù)的用法,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2021-10-10
怎樣使用PowerMockito 測(cè)試靜態(tài)方法
這篇文章主要介紹了使用PowerMockito 測(cè)試靜態(tài)的方法,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2021-07-07
Spring自帶定時(shí)任務(wù)@Scheduled注解實(shí)例講解
這篇文章主要介紹了Spring自帶定時(shí)任務(wù)@Scheduled注解的相關(guān)知識(shí),本文通過實(shí)例代碼給大家介紹的非常詳細(xì),感興趣的朋友跟隨小編一起看看吧2024-06-06

