JUC原始碼學習筆記——執行緒池,FutureTask,Executor框架原始碼解析

一丶Executor框架概覽

剛接觸Java執行緒池的時候,常常被

ThreadPoolExecutor

Executor

ExecutorService

Future

FutureTask

搞得一頭霧水,下面我們先來來理一理這些類的關係。

1。Executor框架的主要結構

任務,表示被執行任務,實現

Runnable

或者

Callable

介面來描述任務需要做什麼事情

任務的執行,包括任務的執行核心介面

Executor

,和實現了

Executor

ExecutorService

介面,其中包括兩個重要的實現

ThreadPoolExecutor

ScheduledThreadPoolExecutor

非同步計算的結果,

Future

表示未來可期,

FutureTask

是其重要實現

2。Executor框架重要類簡介

2。1 Executor 介面

public interface Executor { void execute(Runnable command);}

Executor

的作用的是把任務提交與每個任務將如何執行的機制進行解耦,其中只有一個方法

execute

,但是其實現類,可能是同步的直接呼叫

Runnable#run

,也可能是非同步開啟執行緒執行,並不要求

Executor

的實現必須是非同步,所以沒有一個叫

submit(Callable)

返回非同步計算結果的

Future

方法

2。2 ExecutorService 介面

ExecutorService

實現了

Executor

介面,提供管理和終止的方法,並且可以生成 Future 來跟蹤一個或多個非同步任務的進度的方法,還提供了批次提交任務的方法。

方法

描述

void shutdown()

關閉執行器,如果還有任務沒有執行完,那麼這些任務還會執行,但是不會接受新的任務,如果已經處於關閉狀態還去呼叫此方法,不會有任何效果

List

shutdownNow()

嘗試停止所有正在執行的任務,返回等待執行但未執行的任務,停止執行任務通常是透過呼叫對應執行緒的

interrupt

方法,如果執行緒自己不響應中斷,那麼無濟於事,任務還是會繼續執行

boolean isShutdown()

如果已經被關閉那麼返回true,通常呼叫

shutdown

shutdownNow

後可能存線上程在執行任務,但是還是返回true

boolean isTerminated()

如果所有任務在關閉後都已完成,則返回 true。請注意,除非呼叫了 shutdown 或 shutdownNow,且所以任務都結束了,否則 isTerminated 永遠不會為真

boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException

關閉當前ExcutorServuce後阻塞當前執行緒,直到所有任務都完成執行,或者發生超時,或者當前執行緒被中斷。

Future

submit(Callable

task)

提供一個具備返回值的任務,返回一個

Future

表示是此任務的非同步執行結果。

Future

submit(Runnable task, T result)

submit(Callable)

類似,但是其非同步返回結果在執行完後是引數

result

List

>

invokeAll(Collection

<? extends Callable>

tasks)throws InterruptedException

批次提交一批任務,阻塞直到所有任務都完成或者任務執行失敗或者當前執行緒被中斷

List invokeAll(Collection<? extends Callable > tasks, long timeout, TimeUnit unit) throws InterruptedException;

批次提交一批任務,阻塞直到所有任務都完成或者任務執行失敗或者當前執行緒被中斷,指定的時間超時

T invokeAny(Collection<? extends Callable > tasks) throws InterruptedException,ExecutionException;

提交一批任務,等待其中一個執行完成,或者直到當前執行緒被中斷,返回時會取消沒有執行完的任務

T invokeAny(

Collection<? extends Callable>

tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException

提交一批任務,等待其中一個執行完成,或者直到當前執行緒被中斷,或者等待時間超時,返回時會取消沒有執行完的任務

2。3 Future 介面

Future 表示非同步計算的結果。提供了檢查計算是否完成、等待其完成以及檢索計算結果的方法

方法

描述

boolean cancel(boolean mayInterruptIfRunning)

嘗試取消此任務的執行。如果任務已完成、已被取消或由於某些其他原因無法取消,則此嘗試將失敗。如果成功,並且在呼叫取消時此任務尚未啟動,則此任務不應該執行。如果任務已經開始,則 mayInterruptIfRunning 引數確定是否應該中斷執行該任務的執行緒以嘗試停止該任務。此方法返回後,對 isDone 的後續呼叫將始終返回 true。如果此方法返回 true,則對 isCancelled 的後續呼叫將始終返回 true。

boolean isCancelled()

如果此任務在正常完成之前被取消,則返回 true。返回:如果此任務在完成之前被取消,則為 true

boolean isDone()

如果此任務完成,則返回 true。完成可能是由於正常終止、異常或取消——在所有這些情況下,此方法都將返回 true。

V get() throws InterruptedException, ExecutionException

如果任務沒有執行完,那麼一直阻塞直到任務完成(執行出現異常也是一種完成),或者直到當前執行緒被阻斷

V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException

V get() throws InterruptedException, ExecutionException

的超時等待版本,如果超時還沒有結束任務,那麼丟擲TimeoutException

二丶FutureTask原始碼解析

1。FutureTask 是什麼

根據原始碼上的註釋,FutureTask是一個可取消的非同步計算,它提供了啟動和取消計算,查詢計算是否完成,以及獲取計算結果(只有計算結束了才能獲取到計算結果,否則被阻塞),一旦計算完成那麼不可以重新或者取消計算,除非呼叫

runAndReset

2。FutureTask 的繼承關係

JUC原始碼學習筆記——執行緒池,FutureTask,Executor框架原始碼解析

Future

表示非同步計算的結果,

Runnale

表示一個任務,

RunnableFuture

實現了二者也是一個介面表示的是具備非同步計算結果的任務(沒有結果的Runnable可以視作結果是null)。

3。FutureTask 內部結構

//任務執行的狀態,使用volatile修飾保證執行緒可見性private volatile int state;//任務處於新建private static final int NEW = 0;//任務正在完成,意味著非同步計算結束,但是結果沒有寫回到 outcomeprivate static final int COMPLETING = 1;//任務正常結束,呼叫callable的call沒有丟擲異常private static final int NORMAL = 2;//呼叫callable的call丟擲異常private static final int EXCEPTIONAL = 3;//任務被取消private static final int CANCELLED = 4;//任務正在被中斷,一般是一個瞬態private static final int INTERRUPTING = 5;//任務已經被中斷private static final int INTERRUPTED = 6;//一般是我們的業務邏輯程式碼封裝而成的,(下面會解釋runnable怎麼半)private Callable callable;//記錄非同步計算的結果,如果成功callable的結果,如果丟擲異常那麼記錄的是異常物件//(下面會解釋為啥這裡不需要用volatile修飾)private Object outcome; // non-volatile, protected by state reads/writes//執行當前任務的執行緒private volatile Thread runner;//一個FutureTask可以被多個執行緒呼叫獲取非同步計算結果的方法//如果非同步計算沒有結束那麼這些方法會被阻塞//WaitNode這種資料結果用於儲存阻塞的執行緒們private volatile WaitNode waiters;

3。1 WaitNode

static final class WaitNode { volatile Thread thread; volatile WaitNode next; WaitNode() { thread = Thread。currentThread(); }}

WaitNode 包含等待的執行緒和,指向其他執行緒的next引用

4。FutureTask 的構造方法

public FutureTask(Callable 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}

4。1 透過Executors。callable將Runnable適配為Callable

public static Callable callable(Runnable task, T result) { if (task == null) throw new NullPointerException(); return new RunnableAdapter(task, result);} static final class RunnableAdapter implements Callable { final Runnable task; final T result; RunnableAdapter(Runnable task, T result) { this。task = task; this。result = result; } public T call() { task。run(); return result; } }

這部分就是一個簡單的介面卡模式,沒什麼好說的

4。2 使用this。state = NEW保證this。callable對其他執行緒的可見性

試想如果執行FutureTask的構造方法的時候,發生重排序,

this。callable

的賦值重排序到外部獲取到構造方法生成的FutureTask的後面,並且立馬有另外一個執行緒呼叫了FutureTask的任務執行方法,這時候

this。callable

還來不及賦值,呼叫執行方法丟擲空指標異常。那麼為什麼不用

volatile

修飾

callable

還能保證其可見性暱,能讓原始碼寫上

// ensure visibility of callable

這行註釋暱?

在《 JUC原始碼學習筆記4——原子類,CAS,Volatile記憶體屏障,快取偽共享與UnSafe相關方法 》的學習筆記中,我們說過volatile變數寫具備如下記憶體屏障

JUC原始碼學習筆記——執行緒池,FutureTask,Executor框架原始碼解析

這裡的

store store

屏障防止了

this。callable

的賦值重排序到

this。state = NEW

之後,且後續的store屏障會保證當前執行緒(構造FutureTask的執行緒)工作記憶體會立馬寫回到主記憶體,並讓其他執行緒關於此FutureTask的快取無效,從而保證了

callable

的執行緒可見性。

5。FutureTask#run 執行任務

5。1帶著問題看原始碼

前面我們說過

FutureTask

是一個可取消的非同步計算,也就是說一個執行緒在執行

FutureTask

的時候可能存在其他執行緒併發的呼叫其取消,注意只有沒有完成的任務可以取消,已經完成的任務不可以取消,那麼doug lea是如何實現

FutureTask

完成前後的取消暱

上面所說的取消是

cancel

方法,這個方法有一個布林型別入參

mayInterruptIfRunning

為true的時候會嘗試中斷執行

FutureTask

的執行緒,那麼如果我們業務中邏輯中響應了中斷,FutureTask是怎麼維護任務狀態從

new

INTERRUPTING

然後到

INTERRUPTED

的暱,

FutureTask

提供獲取計算結果的

get

方法,如果FutureTask正在執行,呼叫

get

方法的執行緒將被阻塞,這就意味著,

FutureTask#run

成功或者失敗都要喚醒阻塞的執行緒(多個執行緒都可以調get)

5。2 原始碼分析

public void run() { //如果不是初始,說明有其他執行緒啟動了,或者說其他執行緒取消了任務 那麼不需要執行 //如果是new 但是cas runner失敗了,說明同時有多個執行緒執行此cas,當前執行緒沒有搶過,那麼不能執行此任務, if (state != NEW || !UNSAFE。compareAndSwapObject(this, runnerOffset, null, Thread。currentThread())) return; try { Callable c = callable; //再次校驗下是否為初始狀態 如果不是 說明在當前線從第一個if到此存在其他執行緒取消任務 //任務啟動之前可以取消任務的執行 //====程式碼點1 原始碼解析用===== if (c != null && state == NEW) { V result; //記錄當前任務是否成功執行,如果Callable程式碼寫錯了, //或者說Callable響應中斷,執行的途中被中斷那麼為false boolean ran; try { //業務邏輯執行 result = c。call(); //成功執行 ran = true; } catch (Throwable ex) { //這裡可能是Callable本身程式碼邏輯錯誤異常 也可能是響應中斷丟擲異常 result = null; ran = false; //但是setExcption只會處理程式碼邏輯異常 setException(ex); } if (ran) //設定任務正常執行結果 set(result); } } finally { runner = null; int s = state; //處理中斷 if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); }}

5。2。1只有初始狀態的任務,且只能允許單個執行緒允許當前任務

if (state != NEW || !UNSAFE。compareAndSwapObject(this, runnerOffset, null, Thread。currentThread())) return;

如果任務當前狀態不是new,說明有其他執行緒運行了,或者說其他執行緒取消了任務

如果是new 但是cas runner失敗了,說明同時有多個執行緒執行此cas,當前執行緒沒有搶過,那麼不能執行此任務,這裡使用CAS確保任務不會被其他執行緒再次執行

5。2。2 任務執行(呼叫callalbe#call前)取消任務那麼任務不會允許

程式碼點1

進行了

if (c != null && state == NEW)

的判斷,其中

c!=null

屬於是空指標PTSD,但是

state == NEW

是為了確保,當前執行緒從5。2。1中cas到

程式碼點1

這一段程式碼執行時間內,沒有其他執行緒取消任務。如果存在其他執行緒取消了任務,那麼

state == NEW

就不成立——

任務執行前可以取消任務

5。2。3 記錄任務執行結果setException和set方法

如果執行出現異常

protected

void setException(Throwable t) {

if

(UNSAFE。compareAndSwapInt(

this

, stateOffset, NEW, COMPLETING)) { outcome = t; UNSAFE。putOrderedInt(

this

, stateOffset, EXCEPTIONAL);

// final state

finishCompletion(); } }

正常執行結束

protected

void

set

(V v) {

if

(UNSAFE。compareAndSwapInt(

this

, stateOffset, NEW, COMPLETING)) { outcome = v; UNSAFE。putOrderedInt(

this

, stateOffset, NORMAL);

// final state

finishCompletion(); } }

這兩個方法都差不多,都是上來一個CAS將

state

從new轉變為

COMPLETING

,然後用

outcome

記錄異常或者記錄成功返回值,然後使用

UNSAFE。putOrderedInt

改變

state

,如果是出現異常,那麼設定狀態為

EXCEPTIONAL

,如果正常結束設定為

NORMAL

5。2。3。1 為什麼使用UNSAFE。putOrderedInt為什麼outcome沒有使用volatile修飾

UNSAFE。putOrderedInt

這個方法我在《 JUC原始碼學習筆記4——原子類,CAS,Volatile記憶體屏障,快取偽共享與UnSafe相關方法 》中關於AtomicInteger

lazySet

中說過,Store load屏障可以讓後續的load指令對其他處理器可見,但是需要將其他處理器的快取設定成無效讓它們重新從主記憶體讀取,

putOrderedInt

提供一個store store屏障,然後寫資料,store store是保證

putOrderedInt

之前的普通寫入和

putOrderedInt

的寫入不會重排序,但是不保證下面的volatile讀寫不被重排序,省去了store load記憶體屏障,提高了效能,但是後續的讀可能存在可見性的問題。

putOrderedInt

的store store屏障保證了

outcome

回立即重新整理回主存

//也就是說protected void set(V v) { if (UNSAFE。compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { //outcome是重新整理回主存的,且不會重排序到putOrderedInt後面 //這也是outcome沒有使用volatile修飾的原因之一, //有後續呼叫putOrderedInt方法保證其對其他執行緒的可見性 outcome = v; //state欄位使用putOrderedInt寫入 其他執行緒存在可見性的問題 UNSAFE。putOrderedInt(this, stateOffset, NORMAL); // final state finishCompletion(); }}

state

的執行緒可見性問題是如何解決的——請接著看下去

5。2。3。2 finishCompletion 完成對等待執行緒的喚醒

private void finishCompletion() { // 等待的節點 for (WaitNode q; (q = waiters) != null;) { //==程式碼點1 後續解析使用== 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; //直到一個為null的節點,意味著遍歷結束 if (next == null) break; q。next = null; // unlink to help gc q = next; } //結束 break; } } //鉤子方法 留給我們自己擴充套件 done(); //將任務置為null callable = null; // to reduce footprint}

這裡拿到

waiters

然後進行自旋遍歷所有等待的節點執行緒然後喚醒它們,有意思的點在

程式碼點1

為何這裡要使用CAS更新

waiters

為null暱,因為這裡存線上程A執行完FutureTask呼叫

finishCompletion

的同時執行緒B呼叫get進行等待,呼叫

get

方法進行排隊(排隊時也是CAS設定自己為waiters)這兩個CAS必定有一個成功,有一個失敗

如果A失敗,說明B在A喚醒之前進行排隊,掛起自己,那麼A在自旋喚醒的時候會喚醒B

如果B失敗,那麼說明B在A喚醒之後進行排隊,那麼這時候不需要排隊了,因為任務已經完成了,B只需要進行自旋獲取返回結果即可

5。2。3。3 處理因為取消任務造成的中斷#handlePossibleCancellationInterrupt

在run方法的finally塊中存在

//執行完設定runner為空runner = null;//重新獲取狀態int s = state; //如果是INTERRUPTING 或者INTERRUPTEDif (s >= INTERRUPTING) //handlePossibleCancellationInterrupt handlePossibleCancellationInterrupt(s);

private void handlePossibleCancellationInterrupt(int s) { if (s == INTERRUPTING) //如果是打斷中 那麼等待直到結束打斷 while (state == INTERRUPTING) Thread。yield();

cancel

方法可以選擇傳入true表示,如果任務還在執行那麼呼叫執行任務執行緒的

interrupt

方法進行中斷,如果是呼叫

cancel

的執行緒還沒有結束中斷那麼當前執行的執行緒會讓步,為什麼這麼做,後續講解

cancel

的時候我們再說

5。3。獲取任務執行結果

public V get() throws InterruptedException, ExecutionException { int s = state; //任務為NEW 和 COMPLETING 那麼呼叫那麼會呼叫awaitDone if (s <= COMPLETING) s = awaitDone(false, 0L); //此方法如果發現Future呼叫異常那麼排除異常 return report(s);}

public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { if (unit == null) throw new NullPointerException(); int s = state; //如果狀態小於COMPLETING 則為 NEW 和 COMPLETING 那麼會呼叫awaitDone //如果awaitDone結束的時候返回的狀態還是 NEW or COMPLETING 丟擲中斷異常 if (s <= COMPLETING && (s = awaitDone(true, unit。toNanos(timeout))) <= COMPLETING) throw new TimeoutException(); //此方法如果發現Future呼叫異常那麼排除異常 return report(s);}

get()

表示無限期等待,直到當前執行緒被中斷,

get(long timeout, TimeUnit unit)

則是超時等待,如果等待時間內任務沒有完成那麼丟擲TimeoutException,如果等待過程被中斷那麼丟擲中斷異常。二者都呼叫了

awaitDone(是否超時等待,等待時長)

此方法返回值是此

FutureTask

狀態。

5。3。1 COMPLETING

FutureTask在呼叫run方法正常結束呼叫

set

執行出現異常呼叫

setException

,這兩個方法首先都會使用CAS將任務狀態從

NEW

轉變為

COMPLETING

,然後設定

outcome

的值,然後再設定任務狀態為

NORMAL

(正常完成),

EXCEPTIONAL

出現異常,最後呼叫

finishCompletion

喚醒所有等待當前FutureTask完成的執行緒。

所有說

COMPLETING

意味著當前任務(程式設計師自己定義的業務邏輯)的run方法呼叫完成,但是任務返回還沒寫到

outcome

上,以及還沒喚醒等待的執行緒。但是隻要

outcome

結果賦值後,立馬改變任務狀態為

NORMAL

或者

EXCEPTIONAL

,而不是呼叫完成

finishCompletion

後改變。為什麼暱

執行緒A呼叫

get()無限期等待

改變任務狀態為NORMAL或者EXCEPTIONAL

之前

那麼這時候執行緒A會進入

awaitDone

如果當前FutureTask狀態為new那麼會掛起自己等待任務執行執行緒呼叫

finishCompletion

,如果當前FutureTask狀態為

COMPLETING

那麼呼叫

yield

讓出cpu,而不是掛起自己,因為後續FutureTask執行的執行緒寫回outcome改變狀態為

NORMAL

或者

EXCEPTIONAL

是很快的,也許修改狀態為

NORMAL

或者

EXCEPTIONAL

會導致執行緒A

yield

使用的是

UNSAFE。putOrderedInt

存線上程可見性問題),但是無關緊要

執行緒A呼叫

get()無限期等待

改變任務狀態為NORMAL或者EXCEPTIONAL

之後

說明當前執行緒A“踩點很準”任務一執行完成立馬來了,那麼當前執行緒A或許會

yield

,也許是直接獲取到執行結果

執行緒A呼叫

get(long,timeunit)超時等待

這種方式多一個超時機制,如果等待結束還是發現任務狀態為new,說明程式設計師自己定義的任務業務邏輯還沒有走完,那麼直接超時異常,如果為

COMPLETING

,執行緒會進行讓步,而不是視作任務執行超時

5。3。2 等待直到任務完成awaitDone方法

此方法還還支援超時等待

private int awaitDone(boolean timed, long nanos) throws InterruptedException { //等待結束時間,如果非超時等待,那麼為0 final long deadline = timed ? System。nanoTime() + nanos : 0L; //當前執行緒如果進入等待任務完成佇列,此變數記錄等待節點 WaitNode q = null; //是否入隊(等待任務完成佇列) boolean queued = false; for (;;) { //如果等待的過程中被中斷, //那麼把自己從等待waiters中刪除 //並且丟擲中斷異常 if (Thread。interrupted()) { removeWaiter(q); throw new InterruptedException(); } //讀取state,volatile保證可見性 int s = state; //如果當前大於COMPLETING 說明任務執行完成, //或者取消了,或者由於取消而被中斷 直接返回當前狀態,不需要再等了 if (s > COMPLETING) { //節點執行緒置為null,後續執行任務執行緒喚醒等待執行緒的時候不會喚醒到此執行緒 if (q != null) q。thread = null; return s; } //如果任務正在完成,進行執行緒讓步 //後續FutureTask執行的執行緒寫回outcome改變狀態為NORMAL或者EXCEPTIONAL是很快的, //也許修改狀態為NORMAL或者EXCEPTIONAL會導致執行緒A多yield幾下(使用的是UNSAFE。putOrderedInt存線上程可見性問題),但是無關緊要 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); }}

這個方法分支超多,我們慢慢分析

如果執行緒執行

awaitDone

之前就被中斷,那麼會直接丟擲中斷異常

如果執行緒執行

awaitDone

並且成功被使用

LockSupport。parkNanos(this, nanos)

或者

LockSupport。park(this)

掛起,這時候被中斷那麼會從這兩個方法中返回,繼續自旋,

Thread。interrupted()

為true,會重置中斷標識並且丟擲中斷異常

如果自旋的時候發現任務狀態大於

COMPLETING

說明當前任務執行完成了,或者說任務被取消,或者由於取消已經中斷了,那麼直接返回即可,從這裡返回有三種情況

第一次自旋發現任務完成了

超時等待指定時間結束髮現任務完成了

任務完成的時候被任務執行執行緒喚醒,繼續自旋發現任務完成了

如果自旋的時候發現任務狀態等於

COMPLETING

那麼呼叫

yield

讓出cpu,而不是掛起自己,因為後續FutureTask執行的執行緒寫回outcome改變狀態為

NORMAL

或者

EXCEPTIONAL

是很快的,也許修改狀態為

NORMAL

或者

EXCEPTIONAL

會導致執行緒A多

yield

幾下(使用的是

UNSAFE。putOrderedInt

存線上程可見性問題),但是無關緊要

注意如果超時等待指定時間結束,繼續自旋,如果進入此分支,那麼讓出cpu,再次獲得時間片後,繼續執行,下一次自旋,而不會進入到下面的超時異常分支,也就是說

COMPLETING

意味著任務執行完了,但是在做一些善後工作(寫入任務返回值,喚醒等待執行緒)不會由於此狀態導致超時

q == null

意味著當前執行緒沒有被包裝成WaitNode,當前執行緒也沒有被中斷,任務沒有完成也不是Completing狀態,這時候呼叫構造方法然後繼續自旋

static

final

class

WaitNode

{

volatile

Thread thread;

volatile

WaitNode next; WaitNode() { thread = Thread。currentThread(); } }

!queued

這是在5的基礎上,當前q已經有了節點,但是還沒有進入等待任務完成佇列,下面透過CAS讓當前執行緒入隊

else

if

(!queued) queued = UNSAFE。compareAndSwapObject(

this

, waitersOffset, q。next = waiters, q);

這裡首先

q。next = waiters

讓當前節點的next指向waiters,然後CAS設定waiters為當前節點,也就是說最後入隊的節點使用waiters記錄,使用next串聯每一個等待的執行緒節點,

q。next = waiters

不需要考慮執行緒安全,和AQS中的入隊類似,這是改變當前節點的next引用指向,但是修改waiters需要考慮執行緒安全問題,如果這裡CAS失敗了,那麼

queued

為false 繼續自旋嘗試CAS自己為waiters

掛起當前執行緒

//超時等待

else

if

(timed) {

//需要等待的時間

nanos = deadline - System。nanoTime();

//已經超時

if

(nanos <=

0

L) {

//把自己從waiters等待任務完成佇列中移除

removeWaiter(q);

return

state; }

//掛起指定時間

LockSupport。parkNanos(

this

, nanos); }

//等待直到被中斷或者喚醒

else

LockSupport。park(

this

);

這裡超時部分多一個

removeWaiter

,將自己從等待任務完成佇列中移除,這個方法的執行需要考慮執行緒安全問題,同樣使用自旋+CAS保證執行緒安全,這裡不做過多分析。

5。3。3 report

private V report(int s) throws ExecutionException { Object x = outcome; //如果任務正常結束 if (s == NORMAL) //強轉 return (V)x; //如果任務取消了 或者由於取消被中斷了,丟擲取消異常 if (s >= CANCELLED) throw new CancellationException(); //反之丟擲ExecutionException 包裝 原始的異常 throw new ExecutionException((Throwable)x);}

只有任務正常執行的時候,才會返回結果,如果被取消那麼丟擲取消異常。

5。4 取消任務

取消有一個比較有趣的點,如果取消在任務執行完之前,那麼說明取消成功,後續任務完成呼叫

set

或者

setException

應該是什麼都不做。如果取消在任務執行完之後,那麼取消的這個動作應該失敗,下面我們看下doug lea如果處理這個細節。

//mayInterruptIfRunning 表示需要中斷任務執行執行緒public boolean cancel(boolean mayInterruptIfRunning) { //任務不是初始,或者CAS修改狀態從new 到INTERRUPTING 或者CANCELLED 失敗 //直接返回false if (!(state == NEW && UNSAFE。compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) return false; try { //如果需要中斷 if (mayInterruptIfRunning) { try { Thread t = runner; //執行中斷 if (t != null) t。interrupt(); } finally { // final state //修改狀態為INTERRUPTED UNSAFE。putOrderedInt(this, stateOffset, INTERRUPTED); } } } finally { //喚醒所有等待任務執行的執行緒 finishCompletion(); } return true;}

第一個if

if

(!(state == NEW && UNSAFE。compareAndSwapInt(

this

, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))

return

false

如果

state == NEW

不成立,說明任務在執行此判斷之前已經結束了(Completing,或者已經到了NORMAL,或者EXCEPTIONAL)說明取消在任務結束之前,那麼直接返回false。或者說當前執行緒A對任務的取消在其他執行緒B取消任務之後,這時候就A執行緒取消返回false

如果

UNSAFE。compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED))

不成立

意味著

state == NEW

判讀的時候成立,但是執行這句CAS的時候之前有老六執行緒搶先一步,或者說存在併發當前執行緒沒有搶過,那麼也直接返回false,這裡保證了

cancel

的執行是序列的,不存線上程安全問題。注意這裡如果需要中斷任務執行執行緒那麼CAS修改狀態到

INTERRUPTING

,反之直接修改到

CANCELLED

嘗試中斷任務執行執行緒

if

(mayInterruptIfRunning) {

try

{ Thread t = runner;

if

(t !=

null

) t。interrupt(); }

finally

{

// final state

UNSAFE。putOrderedInt(

this

, stateOffset, INTERRUPTED); } }

呼叫

interrupt

具體如何響應中斷,得看程式設計師定義的業務邏輯是啥樣的。呼叫putOrderedInt修改狀態為INTERRUPTED,表示已經完成了中斷

喚醒等待任務結束的執行緒

直接呼叫

finishCompletion

,這個方法前面分析過。這裡假如執行緒A取消了任務,那麼執行緒B任務執行完後呼叫

set

或者

setException

會如何暱——什麼都不做

protected

void

set

(V v) {

//此時state 是INTERRUPTING 或者INTERRUPTED if為false

if

(UNSAFE。compareAndSwapInt(

this

, stateOffset, NEW, COMPLETING)) { outcome = v; UNSAFE。putOrderedInt(

this

, stateOffset, NORMAL);

// final state

finishCompletion(); } }

run方法對中斷的處理

public

void

run

() {

if

(state != NEW || !UNSAFE。compareAndSwapObject(

this

, runnerOffset,

null

, Thread。currentThread()))

return

try

{

//省略任務的執行

}

finally

{ runner =

null

int

s = state;

//進入這個if 必須是INTERRUPTING,或者INTERRUPTED

if

(s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } }

private

void

handlePossibleCancellationInterrupt

int

s) {

if

(s == INTERRUPTING)

while

(state == INTERRUPTING) Thread。

yield

(); }

這裡為INTERRUPTING ,可能是取消任務的執行緒還沒來得及執行

UNSAFE。putOrderedInt(this, stateOffset, INTERRUPTED)

,也可能是可見性導致執行任務的執行緒沒有讀取到最新的state,

handlePossibleCancellationInterrupt

會讓執行任務的執行緒等待,這點我不是很理解其必要性在哪裡,也許回了確保任務執行執行緒在執行完

FutureTask

的run方法之後,一定已經被中斷了。

三丶ThreadPoolExecutor 原始碼解析

1。執行緒池概覽

1。1執行緒池是什麼

一個

ExecutorService

的實現,會使用池中的一個執行緒處理提交的任務。池中存在n個執行緒,當前任務提交時,可能會new一個執行緒執行執行任務,也可能讓當前任務進行排隊。執行緒池(Thread Pool)是一種基於池化思想管理執行緒的工具,經常出現在多執行緒伺服器中。

1。2 執行緒池的結構

執行緒池可以看作兩部分,執行工作的工人和在佇列中等待被執行的任務

2丶為什麼要使用執行緒池,執行緒池解決了什麼問題

2。1。使用執行緒池的好處

降低資源消耗

:透過池化技術重複利用已建立的執行緒,降低執行緒建立和銷燬造成的損耗。java執行緒和作業系統時一對一的對映關係,新建或者銷燬一個執行緒都存在資源的消耗

提高響應速度

:任務到達時,一定情況下無需等待執行緒建立即可立即執行。

提高執行緒的可管理性

:執行緒是稀缺資源,如果無限制建立,不僅會消耗系統資源,還會因為執行緒的不合理分佈導致資源排程失衡,降低系統的穩定性。使用執行緒池可以進行統一的分配、調優和監控。使用過多的執行緒會導致執行緒上下文切換更多,從而導致在儲存“現場”和恢復“現場”的開銷激增

提供更多更強大的功能

:執行緒池具備可拓展性,允許開發人員向其中增加更多的功能。比如延時定時執行緒池ScheduledThreadPoolExecutor,就允許任務延期執行或定期執行

2。2。執行緒池解決了什麼問題

頻繁申請/銷燬資源和排程資源,將帶來額外的消耗,可能會非常巨大。

所以使用執行緒池,儲存執行緒,避免每到一個任務需要新建一個執行緒,也避免執行完任務之後執行緒立馬消亡

對資源無限申請缺少抑制手段,易引發系統資源耗盡的風險。

執行緒池可以設定最大執行緒數,執行緒池持有的執行緒不可超過此值

系統無法合理管理內部的資源分佈,會降低系統的穩定性。

3。執行緒池ThreadPoolExecutor的繼承關係

JUC原始碼學習筆記——執行緒池,FutureTask,Executor框架原始碼解析

Executor

ExecutorService

兩個介面前面分析了,下面看下

AbstractExecutorService

3。1AbstractExecutorService

AbstractExecutorService

提供了

Runnable

Callable

適配成

RunnableFuture

(一般適配成FutureTask),還實現了

ExecutorService

submit

invokeAny

,以及

invokeAll

。是對

ExecutorService

的抽象實現,有點模板方法的意思

3。1。1Runnable,Callable適配

protected RunnableFuture newTaskFor(Runnable runnable, T value) { return new FutureTask(runnable, value);}

rotected RunnableFuture newTaskFor(Callable callable) { return new FutureTask(callable);}

這裡都new了一個

FutureTask

,其中適配

Runnable

呼叫了

Executors#callable

方法,最後new了一個

RunnableAdapter

,典型的介面卡模式

static final class RunnableAdapter implements Callable { final Runnable task; final T result; RunnableAdapter(Runnable task, T result) { this。task = task; this。result = result; } public T call() { task。run(); return result; }}

3。1。2 submit

public Future submit(Callable task) { if (task == null) throw new NullPointerException(); RunnableFuture ftask = newTaskFor(task); execute(ftask); return ftask; }//省略了submit(Runnable)和 submit(Runnable,T)

最後都是呼叫的

execute(ftask)

方法,此方法有具體子類實現,

3。1。3 執行任意一個 invokeAny

執行任意一個的難點在於,只要有任何一個執行完成了那麼就需要返回,而不是阻塞等待所有任務完成,但是如果一直遍歷所有任務執行狀態是否是完成,那麼會消耗CPU,那麼douglea是如何實現的暱?

invokeAny

方法最終呼叫

doInvokeAny

進行實現,具體的做法是將提交的任務,包裝成

ExecutorCompletionService的內部類QueueingFuture

private class QueueingFuture extends FutureTask { QueueingFuture(RunnableFuture task) { super(task, null); this。task = task; } //任務完成之後會回撥 done方法 這時候, //執行執行緒會把當前任務放入到 completionQueue——完成佇列之中 protected void done() { completionQueue。add(task); } //原來的任務 private final Future task;}

completionQueue

是一個BlockingQueue 通常是無解阻塞佇列

LinkedBlockingQueue

,所有進行自旋遍歷的時候,會由於佇列中沒有完成的任務而阻塞掛起,減少CPU資源的消耗,而多個執行緒在執行任務,只要任務結束了(任務執行失敗,被取消也是結束,正常完成也是結束)就會回撥

done

方法把當前任務放入到

completionQueue

等待的執行緒將被喚醒,獲取到AQS鎖的執行緒將呼叫take獲取最先完成的任務

然後釋放鎖(這部參閱我的 JUC原始碼學習筆記3——AQS等待佇列和CyclicBarrier,BlockingQueue )接著其他的執行緒也如此往復,take方法是會刪除等待佇列頭部的,也就是說多個執行緒呼叫

invokeAny

那怕任務不一樣也是獲取到不同的執行結果,3個任務被,四個執行緒呼叫

invokeAny

,最後拿到鎖的執行緒拿到空

3。1。4 執行所有 invokeAll

invokeAll

相比於

invokeAny

就簡單許多,直接迴圈執行每一個任務,然後呼叫

Future#get

方法就好,get方法會讓當前執行緒阻塞,也就是說任務執行時間>Max(耗時最長任務時間),

invokeAny

則是>min(任務時間),但是

invokeAll

返回

List>

,並且忽略單個任務的超時和取消丟擲的異常,如果存在任何一個任務執行被中斷且丟擲中斷異常,那麼會去取消還在執行的任務並且進行中斷,invokeAll還有一個超時版本,如果超時會直接返回所有Future。

invokeAny

返回的是

T

也就是任務執行結果,而不是

Future

並且不忽略取消和中斷也不忽略任務執行失敗丟擲的異常

3。2 執行緒池ThreadPoolExecutor的狀態和屬性

3。2。1 狀態和狀態的變更

具體變更原始碼會在後續講到

//ctl 高三位標識執行緒池的狀態 低29位標識執行緒池工作執行緒個數//一個int 標識方便了在更新狀態和工作執行緒個數不需要加鎖private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//29 移位操作使用private static final int COUNT_BITS = Integer。SIZE - 3;//2的29次方減1 標識執行緒池工作執行緒個數的最大值private static final int CAPACITY = (1 << COUNT_BITS) - 1;//RUNNING:接受新任務並處理排隊任務,private static final int RUNNING = -1 << COUNT_BITS;// SHUTDOWN:不接受新任務,但處理排隊任務private static final int SHUTDOWN = 0 << COUNT_BITS;//STOP:不接受新任務,不處理排隊任務,並中斷正在進行的任務private static final int STOP = 1 << COUNT_BITS;// TIDYING:所有任務都已終止,workerCount 為零private static final int TIDYING = 2 << COUNT_BITS;// TIDYING 狀態後將執行 terminate() 鉤子方法。然後來到TERMINATED狀態private static final int TERMINATED = 3 << COUNT_BITS;//儲存待處理任務的阻塞佇列private final BlockingQueue workQueue;//鎖,執行緒池用一個set儲存所有執行緒,一個int儲存最大的執行緒數,修改這些的時候使用這個鎖private final ReentrantLock mainLock = new ReentrantLock();//包含池中所有工作執行緒的集合。僅在持有 mainLock 時訪問private final HashSet workers = new HashSet();//呼叫awaitTermination的執行緒在此等待佇列上等待private final Condition termination = mainLock。newCondition();//執行緒池中存在的最大的工作執行緒數。只能在 mainLock 下訪問。private int largestPoolSize;//已完成任務的計數器。僅在工作執行緒終止時更新。只能在 mainLock 下訪問。private long completedTaskCount;//新執行緒的工廠。所有執行緒都是使用這個工廠建立的private volatile ThreadFactory threadFactory;//拒絕策略,當達到最大執行緒數,並且佇列也無法容納任務的時候呼叫此方法private volatile RejectedExecutionHandler handler;//工作執行緒多久(納秒)沒有執行任務將被回收,(一般針對非核心執行緒,也可以用於核心執行緒的回收)private volatile long keepAliveTime;//如果為 false(預設),核心執行緒即使在空閒時也保持活動狀態。如果為真,核心執行緒使用 keepAliveTime 超時等待工作。private volatile boolean allowCoreThreadTimeOut;//核心執行緒數,如果池中執行緒數小於核心執行緒數,那麼接受新任務總是new一個執行緒private volatile int corePoolSize;//最大執行緒數,當池中執行緒數達到此值,將會把新任務放置在阻塞佇列private volatile int maximumPoolSize;//預設拒絕的策略,丟擲拒絕異常private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

3。3執行緒池原始碼分析

3。3。1 執行任務execute

JUC原始碼學習筆記——執行緒池,FutureTask,Executor框架原始碼解析

public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl。get(); //如果當前工作執行緒總數小於核心執行緒 if (workerCountOf(c) < corePoolSize) { //那麼會嘗試新增一個核心執行緒執行當前任務 //如果新增任務成功那麼直接返回 if (addWorker(command, true)) return; //新增失敗那麼重新獲取執行緒總數和執行緒池狀態 //新增失敗可能是併發提交任務,當前執行緒嘗試增加的時候已經達到核心執行緒數 //或者嘗試新增執行緒的時候,其他執行緒關閉了執行緒池 c = ctl。get(); } //如果是執行狀態 且加入到了任務佇列 //上面新增核心執行緒失敗,但是滿足執行緒池是執行狀態說明是併發提交任務核心執行緒數已經達到了 if (isRunning(c) && workQueue。offer(command)) { //如果新增成功 重新獲取程總數和執行緒池狀態 int recheck = ctl。get(); //如果發現不行執行狀態了嘗試刪除任務,說明加入後執行緒池被關了 //isRunning(c) && workQueue。offer(command) 不是一個原子操作 if (!isRunning(recheck) && remove(command)) //如果成功從佇列刪除了任務,那麼呼叫拒絕策略 reject(command); //反之 新增一個非核心執行緒,儘可能保證佇列中的任務會被執行 else if (workerCountOf(recheck) == 0) addWorker(null, false); } //如果佇列滿了,或者說不是running 那麼新增一個非核心執行緒 //不是running 新增非核心執行緒也不是一定成功的 //如果新增非核心失敗 那麼呼叫拒絕策略 else if (!addWorker(command, false)) reject(command);}

這裡新增工作執行緒呼叫了

addWorker(執行的任務,是否核心執行緒)

if (workerCountO(c) < corePoolSize)

這個分支說明了,當執行緒池中執行緒數沒有達到核心執行緒數的時候都是無腦增加核心執行緒,但是呼叫

addWorker

增加執行緒的時候是要獲取全域性鎖

mainLock

的。如果新增核心執行緒成功了那麼直接返回,反之會重新獲取以下執行緒池的狀態,這裡失敗,可能

if (workerCountO(c) < corePoolSize)

if (addWorker(command, true))

不是一個原子操作,中間可能被其他執行緒更改了執行緒池執行狀態,或者說多執行緒併發呼叫

addWorker

增加核心執行緒數,核心執行緒達到了

corePoolSize

if (isRunning(c) && workQueue。offer(command))

來到這裡可能時新增核心執行緒失敗,或者說一進來就已經達到了核心執行緒數,這裡先檢查時執行狀態,然後向佇列塞任務,這裡的。同樣的

if (isRunning(c) && workQueue。offer(command))

也不是一個原子操作,可能塞的一瞬間執行緒池不是running狀態了,或者說任務阻塞佇列滿了(來到下面的3)。如果不是running了但是阻塞佇列接受了任務,會進入下面的判斷,重新檢查執行緒池的狀態如果不是running那麼嘗試刪除任務,如果成功刪除了任務,說明這個任務沒有被工作執行緒拿去執行,這時候直接呼叫拒絕策略。

如果佇列滿了,或者說不是running了,來到

else if (!addWorker(command, false))

雖然不是running了能不能新增非核心執行緒

addWorker

方法會控制的,如果佇列滿了,會嘗試新增非核心執行緒,這時候是奔著最大執行緒數去的

3。3。2 新增工作執行緒addWorker

private boolean addWorker(Runnable firstTask, boolean core) { retry: //=========自旋部分開始================= for (;;) { int c = ctl。get(); int rs = runStateOf(c); //如果大於等於SHUTDOWN 且 執行緒池不是SHUTDOWN 說明是STOP TIDYING TERMINATED這幾種都是不接受新任務的 //大於等於SHUTDOWN 且佇列是空,這時候也不接受新任務,執行緒池即將關閉 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue。isEmpty())) return false; //自選 for (;;) { int wc = workerCountOf(c); //如果大於(2^29)-1 直接不可新增執行緒,ctl 高三位狀態低29位執行緒數 再多表示不了了 // 如果表示新增核心執行緒 且大於核心執行緒 或者非核心但是大於最大執行緒數 返回false if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //cas增加 工作執行緒數 這裡只是更新ctl 不是真的增加一個執行緒 //這樣增加成功了才能退出 保證了執行緒數不會超過閾值 if (compareAndIncrementWorkerCount(c)) break retry; //如果增加失敗了重新看下狀態,狀態改變了,那麼重新自旋 //cas失敗了,狀態沒變也會自選 c = ctl。get(); // Re-read ctl if (runStateOf(c) != rs) continue retry; } } //=========自旋部分結束================= boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { //新建一個執行緒 w = new Worker(firstTask); final Thread t = w。thread; if (t != null) { //上鎖 final ReentrantLock mainLock = this。mainLock; mainLock。lock(); try { //執行緒池狀態 int rs = runStateOf(ctl。get()); //如果小於 SHUTDOWN 說明是RUNNING //或者是SHUTDOWN 但是沒有任務執行,說明是為了執行佇列中的任務或者預熱執行緒池 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t。isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); //加到set集合 workers。add(w); int s = workers。size(); //更新最大執行緒數 if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { //解鎖 mainLock。unlock(); } //如果加入了set 啟動worker if (workerAdded) { t。start(); workerStarted = true; } } } finally { //如果沒有啟動 說明執行緒池已經不接受新任務了,或者其他奇奇怪怪的異常 if (! workerStarted) //嘗試減少工作執行緒數 並且嘗試關閉執行緒池 addWorkerFailed(w); } //返回worker是否啟動了 return workerStarted;}

自旋部分

這部分的目的是保證池中執行緒數不超過核心執行緒數(如果新增核心執行緒),最大執行緒數(如果新增非核心執行緒)

前置判斷

if (

rs

>= SHUTDOWN

&&

! (

rs

== SHUTDOWN

&&

firstTask == null

&&

! workQueue。isEmpty())) return false

1。STOP TIDYING TERMINATED這幾種狀態不接受新任務

2。SHUTDOWN 狀態但是提交新任務,直接返回false,SHUTDOWN 會處理佇列的任務但是不接受新任務

3。SHUTDOWN 狀態且佇列是空,這時候也不接受新任務,執行緒池即將關閉

內部自旋自增工作執行緒數

for (

;;) {

int wc = workerCountOf(

c

if (

wc

>= CAPACITY || wc >= (

core

? corePoolSize : maximumPoolSize)) return false

if (

compareAndIncrementWorkerCount

c

)) break retry

c = ctl。get()

; // Re-read ctl

if (

runStateOf

c

) != rs) continue retry

// else CAS failed due to workerCount change

; retry inner loop

}

首先沒有新增執行緒之前的工作執行緒數, 不能大於等於最大容量,如果準備新增核心執行緒那麼不能大於等於

corePoolSize

,如果準備新增非核心執行緒那麼不能大於等於

maximumPoolSize

,然後來一個cas

if (compareAndIncrementWorkerCount(c))

,如果CAS表示成功增加了執行緒數,但是沒有增加執行緒,相當於佔了一個坑。那啥時候會失敗暱——併發增加,cas失敗,這時候會繼續自旋。接著

if (runStateOf(c) != rs)

如果發現狀態改變了

continue retry

從外層for開始執行,這樣會重新判斷執行緒狀態,反之從內層for執行,判斷執行緒數併產生重新增加。

這一段自旋相當於一個樂觀鎖,當多執行緒一起提交任務的時候,大家都會進行自旋嘗試增加工作執行緒數,doug lea不是在程式入口加排他鎖,而是使用樂觀鎖,是因為後面還有new一個工作執行緒等操作,只需要樂觀鎖保證數目,new的部分是可以併發的。

增加工作執行緒

在new完worker之後加鎖將worker放入

workers

中,並更新最大執行緒數,然後解鎖,再嘗試啟動工作執行緒,這裡可以看到dougLea儘可能把鎖的粒度控制到最小,new一個工作執行緒和啟動工作執行緒是可以併發的,所有沒有加鎖。如果沒有成功啟動工作執行緒會進入

addWorkerFailed

方法

private

void

addWorkerFailed

(Worker w) { final ReentrantLock mainLock =

this

。mainLock; mainLock。

lock

();

try

{

if

(w !=

null

) workers。

remove

(w); decrementWorkerCount(); tryTerminate(); }

finally

{ mainLock。unlock(); } }

這裡從工作執行緒集合中刪除工作執行緒,自旋cas減少工作執行緒數目,嘗試關閉執行緒池(這個方法內部會判斷執行緒池狀態,並不是嘗試關就一定會關)這一步就是上面操作的回滾

3。3。3 Worker工作執行緒內部類

JUC原始碼學習筆記——執行緒池,FutureTask,Executor框架原始碼解析

Worker這個類繼承了AQS實現了Runnable介面,繼承Runnalbe 比較好理解,比較Worker的職責就是從阻塞佇列中不斷獲取任務執行。但是為什麼Worker為什麼要繼承AQS暱?繼承AQS是為了處理中斷,如果是由於獲取阻塞佇列任務而掛起的話,這時候是需要響應中斷的,但是如果是正在執行任務的話,是不會響應中斷的,這是一個策略問題。

1。初始化時設定state=-1只有在啟動Worker後才會響應中斷

Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this。firstTask = firstTask; this。thread = getThreadFactory()。newThread(this);}

void interruptIfStarted() { Thread t; //只有state 大於等於才會響應中斷 if (getState() >= 0 && (t = thread) != null && !t。isInterrupted()) { try { t。interrupt(); } catch (SecurityException ignore) { } }}

2。Worker#run

public void run() { runWorker(this);}

final void runWorker(Worker w) { Thread wt = Thread。currentThread(); Runnable task = w。firstTask; w。firstTask = null; /// 這裡會將 state置為0 interruptIfStarted 才可以中斷 w。unlock(); boolean completedAbruptly = true; try { //從佇列中獲取任務 如果getTask的時候中斷,不會中斷當前執行緒獲取任務 while (task != null || (task = getTask()) != null) { //上鎖 不響應中斷 w。lock(); //如果是STOP 或TIDYING 或TERMINATED 確保當前執行緒執行緒會被中斷 //如果池正在停止,請確保執行緒被中斷; //如果沒有,請確保執行緒不被中斷。 if ((runStateAtLeast(ctl。get(), STOP) || (Thread。interrupted() && runStateAtLeast(ctl。get(), STOP))) && !wt。isInterrupted()) wt。interrupt(); try { //鉤子方法 beforeExecute(wt, task); Throwable thrown = null; try { //執行 task。run(); } catch (RuntimeException x) { thrown = x; throw x; } catch (Error x) { thrown = x; throw x; } catch (Throwable x) { thrown = x; throw new Error(x); } finally { //鉤子方法 afterExecute(task, thrown); } } finally { task = null; w。completedTasks++; w。unlock(); } } //執行任務時丟擲異常那麼此為true 反之則為false completedAbruptly = false; } finally { //進入這裡可能是執行業務邏輯錯誤,也可能時佇列沒有任務了 processWorkerExit(w, completedAbruptly); }}

關於中斷

lock方法一旦獲取了獨佔鎖,表示當前執行緒正在執行任務中,如果正在執行任務,則不應該中斷執行緒。

如果該執行緒現在不是獨佔鎖的狀態,也就是空閒的狀態,說明它沒有在處理任務,這時可以對該執行緒進行中斷。

執行緒池在執行shutdown方法或tryTerminate方法時會呼叫interruptIdleWorkers方法來中斷空閒的執行緒,interruptIdleWorkers方法會使用tryLock方法來判斷執行緒池中的執行緒是否是空閒狀態;如果執行緒是空閒狀態則可以安全回收。

獲取任務的時候不會由於中斷而丟擲中斷異常退出自旋

這部分邏輯在

getTask

方法中,後面我們討論

如果池停止了那麼確保當前執行緒一定會被中斷,如果池沒有停止那麼確保當前執行緒不被中斷。(shutdownNow和shutdown會中斷執行緒)

if

((runStateAtLeast(ctl。

get

(),

STOP

) || (Thread。interrupted() && runStateAtLeast(ctl。

get

(),

STOP

))) && !wt。isInterrupted()) wt。interrupt();

(runStateAtLeast(ctl。get(), STOP)&&!wt。isInterrupted()

成立

說明這個時候池狀態至少STOP,這種情況下,拿到了佇列中的任務還是是需要處理隊中的任務的,所有這裡這是中斷當前執行緒

wt。interrupt()

,具體怎麼應對中斷這取決於業務邏輯程式碼。

(Thread。interrupted() && runStateAtLeast(ctl。get(), STOP))) &&!wt。isInterrupted()

這依據非常巧妙,如果成立了,說明池停止了,但是

Thread。interrupted()

重置了中斷標識這時候

!wt。isInterrupted()

就是true,那麼會執行

wt。interrupt()

補上中斷標識。如果不成立,說明池沒有停止那麼

Thread。interrupted()

復位了中斷標識,保證了池沒有停止的時候執行緒沒有被中斷

回收執行緒

如果執行緒執行業務程式碼丟擲異常,或者說佇列中沒有任務了,或者說當前執行緒空閒太久沒有執行任務,會回收工作執行緒

private void processWorkerExit(Worker w, boolean completedAbruptly) { //執行業務邏輯錯誤 那麼自旋減少工作執行緒數 if (completedAbruptly) // If abrupt, then workerCount wasn‘t adjusted decrementWorkerCount(); final ReentrantLock mainLock = this。mainLock; mainLock。lock(); try { //完成任務資料加到執行緒池完成任務數上 completedTaskCount += w。completedTasks; //回收當前執行緒 workers。remove(w); } finally { mainLock。unlock(); } //嘗試關閉執行緒池,會中斷一個工作執行緒,“傳播”關閉訊號,執行狀態的執行緒池不會受影響 tryTerminate(); int c = ctl。get(); //SHUTDOWN RUNNING 狀態 if (runStateLessThan(c, STOP)) { //當前執行緒不是執行任務業務邏輯錯誤 if (!completedAbruptly) { //如果執行核心執行緒回收那麼為 最小為0 反之執行緒池保證最起碼有核心執行緒數個執行緒 int min = allowCoreThreadTimeOut ? 0 : corePoolSize; // 佇列還有任務那麼流一個執行緒 if (min == 0 && ! workQueue。isEmpty()) min = 1; //如果大於最小 那麼結束 if (workerCountOf(c) >= min) return; // replacement not needed } //如果是業務異常退出 //或者不是業務邏輯就是沒有拿到任務,但是執行緒池個數少於min那麼新增一個工作執行緒 addWorker(null, false); }}

3。3。4 從阻塞佇列中獲取任務getTask

private Runnable getTask() { //獲取任務是否超時 boolean timedOut = false; for (;;) { int c = ctl。get(); int rs = runStateOf(c); // 如果執行緒池為STOP TIDYING TERMINATED 那麼cas減小執行緒數 return //如果SHUTDOWN 但是佇列存在任務 不會cas減少,那麼不會return if (rs >= SHUTDOWN && (rs >= STOP || workQueue。isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); //如果允許核心執行緒超時被回收 那麼為true 或者工作執行緒大於核心執行緒數會沒有任務的時候會減少到核心執行緒數 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; //如果工作執行緒大於最大核心數 或者 允許過期且獲取任務超時 if ((wc > maximumPoolSize || (timed && timedOut)) //如果佇列不是空至少保證wc大於1 那麼減少後工作執行緒至少為1 && (wc > 1 || workQueue。isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; //如果CAS失敗那麼繼續自旋 continue; } try { //如果允許核心執行緒超時,且當前工作執行緒小於核心執行緒數 那麼會使用響應中斷但是無限等待的take方法,反之使用超時等待的pool Runnable r = timed ? workQueue。poll(keepAliveTime, TimeUnit。NANOSECONDS) : workQueue。take(); if (r != null) //獲取到任務 那麼直接返回任務 return r; //反之說明超時沒有獲取到任務 timedOut = true; } catch (InterruptedException retry) { //如果被中斷那麼把超時置為false 繼續自旋 timedOut = false; } }

這裡我們可以看到

getTask

是不會響應中斷的,如果在

poll

或者

take

的時候被中斷會設定

timedOut=fasle

然後繼續自旋,也就是說

getTask

返回要麼是拿到了任務,要麼是超時沒有獲取到任務說明當前執行緒空閒需要被回收,但是這個空閒需要被回收有一個限制——要麼允許核心執行緒空閒被回收,要麼當前工作執行緒數大於核心執行緒數。這裡對中斷的處理的目的是,那怕執行緒池被shutdown不會中斷

getTask

的執行緒,因為

shutdown

需要確保佇列中的任務要被處理完。

3。3。5 shutdown,shutDownNow

public void shutdown() { final ReentrantLock mainLock = this。mainLock; mainLock。lock(); try { checkShutdownAccess(); //確保狀態至少為SHUTDOWN,因為終止的同時新增執行緒數的可能這裡使用自旋+cas advanceRunState(SHUTDOWN); //中斷所有的空閒工作執行緒,會首先獲取鎖,保證工作執行緒是空閒才會中斷,如果正在執行說明在處理佇列中的任務不會中斷工作執行緒,這裡沒有中斷一個空閒的執行緒是為了讓多餘的worker就會立即退出,而不是等待一個落後的任務完成 interruptIdleWorkers(); //鉤子函式 onShutdown(); } finally { mainLock。unlock(); } //嘗試終止執行緒池 tryTerminate();}public List shutdownNow() { List tasks; final ReentrantLock mainLock = this。mainLock; mainLock。lock(); try { checkShutdownAccess(); advanceRunState(STOP); //注意這裡 shutdown是interruptIdleWorkers //這個方法比較無腦直接中斷中斷執行緒 interruptWorkers(); //不同點在於 這兒 這裡會把所有未完成的任務放到集合中返回 tasks = drainQueue(); } finally { mainLock。unlock(); } tryTerminate(); return tasks;}

shutdown

關閉執行緒池,但是會讓工作執行緒處理完佇列中任務,但是

shutdownNow

會呼叫

drainQueue

排空工作佇列中的任務,也就是說被排空的任務不會被執行。

3。3。6 interruptIdleWorkers,interruptWorkers 和 interruptIfStarted 中斷worker

interruptIdleWorkers

worker。tryLock

成功才會中斷工作執行緒,tryLock成功意味著worker處於空閒狀態(worker一旦啟動就會呼叫lock方法,表示處於工作狀態)

private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this。mainLock; mainLock。lock(); try { for (Worker w : workers) { Thread t = w。thread; //注意這裡,只有worker沒有被中斷,且 嘗試上鎖成功後才會中斷worker //tryLock 成功那麼worker的獨佔鎖被獲取了,後續worker就無法運行了 if (!t。isInterrupted() && w。tryLock()) { try { t。interrupt(); } catch (SecurityException ignore) { } finally { w。unlock(); } } if (onlyOne) break; } } finally { mainLock。unlock(); }}

interruptWorkers 中斷所有啟動的工作執行緒,無論是否空閒

private void interruptWorkers() { final ReentrantLock mainLock = this。mainLock; mainLock。lock(); try { //獲取全域性鎖迴圈遍歷每一個worker,這裡全域性鎖的獲取可以避擴音交任務是新建執行緒加入到workers 集合中 for (Worker w : workers) w。interruptIfStarted(); } finally { mainLock。unlock(); }}

void interruptIfStarted() { Thread t; //這裡判斷了state大於等於0 才能中斷,初始化worker的時候置為-1 只有worker啟動後才可以中斷 if (getState() >= 0 && (t = thread) != null && !t。isInterrupted()) { try { t。interrupt(); } catch (SecurityException ignore) { } }}

3。3。7 嘗試終止執行緒池tryTerminate

final void tryTerminate() { for (;;) { int c = ctl。get(); //如果是執行狀態 或者已經為TIDYING或者 TERMINATED //或者是SHUTDOWN 但是佇列還有任務需要執行那麼直接返回 if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue。isEmpty())) return; //不符合上面if 且工作執行緒不為0 那麼中斷其中一個,說明此時為shutdown,且佇列為空 //後續這個被中斷的執行緒會中斷後續的工作執行緒 if (workerCountOf(c) != 0) { // Eligible to terminate interruptIdleWorkers(ONLY_ONE); return; } //鎖 final ReentrantLock mainLock = this。mainLock; mainLock。lock(); try { //設定為TIDYING 工作執行緒數為0 if (ctl。compareAndSet(c, ctlOf(TIDYING, 0))) { try { //呼叫terminated 鉤子方法 terminated(); } finally { //設定成工作執行緒數為0 狀態為TERMINATED ctl。set(ctlOf(TERMINATED, 0)); //喚醒在termination 上等待的執行緒 //如果呼叫awaitTermination會阻塞直到執行緒池關閉 termination。signalAll(); } return; } } finally { mainLock。unlock(); } // else retry on failed CAS }

這裡可以看到,如果為shutdown,但是佇列還有任務讓佇列的任務得到處理。這裡比較不好理解的是

interruptIdleWorkers(ONLY_ONE)

中斷一個執行緒,為什麼說中斷一個執行緒後續會傳播到其他執行緒暱,如果在shutdown呼叫

tryTerminate

執行中斷一個執行緒的這一瞬間,這個執行緒在

getTask

那麼不好有啥影響,他會繼續處理佇列中的任務(如果是shutdownNow,會把阻塞中的任務排空,那麼當前這個執行緒可能拿不到任務,也可能在排空之前拿到任務繼續執行)

getTask如果為空 那麼會執行

processWorkerExit

後續還是會呼叫

tryTerminate

,如果

getTask

不為null,執行任務的途中發現當前執行緒池處於關閉狀態為設定當前執行緒中斷表示,然後繼續允許任務的業務邏輯,是否丟擲中斷異常取決業務邏輯程式碼,如何丟擲異常那麼也會執行

processWorkerExit

後續還是會呼叫

tryTerminate

,如果沒有丟擲異常那麼繼續

getTask

,一進來直接判斷是否處於關閉,如果是Stop及以上那麼直接減少執行緒數返回null會首先從

如果shutdown會保證佇列中的任務被執行完,後續執行完還是會呼叫

processWorkerExit

中的

tryTerminate

3。3。8 阻塞直到執行緒池被終止,或者被中斷,或者超時

public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { long nanos = unit。toNanos(timeout); final ReentrantLock mainLock = this。mainLock; mainLock。lock(); try { for (;;) { //如果呼叫此方法的時候已經是TERMINATED 返回ture if (runStateAtLeast(ctl。get(), TERMINATED)) return true; //如果等待時間小於0 返回false if (nanos <= 0) return false; //超時等待 中斷丟擲中斷異常 超時丟擲超時異常 nanos = termination。awaitNanos(nanos); } } finally { mainLock。unlock(); }}

這裡說的執行緒池被終止是指執行完shutdown,shutdownNow,也就是說執行完tryTerminate中的

terminated

並且設定執行緒池為TERMINATED,後續才會被喚醒,這裡涉及到AQS的等待佇列—— JUC原始碼學習筆記3——AQS等待佇列和CyclicBarrier,BlockingQueue

3。3。9 拒絕策略

CallerRunsPolicy 由提交任務的執行緒執行任務,如果執行緒池是STOP,TIDYING,TERMINATED 那麼一聲不吭的拋棄任務

public static class CallerRunsPolicy implements RejectedExecutionHandler { public CallerRunsPolicy() { } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { //如果不是RUNNING if (!e。isShutdown()) { r。run(); } }}

AbortPolicy 直接丟擲RejectedExecutionException

public static class AbortPolicy implements RejectedExecutionHandler { public AbortPolicy() { } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { throw new RejectedExecutionException(“Task ” + r。toString() + “ rejected from ” + e。toString()); }}

DiscardPolicy 悄無聲息的忽略任務 什麼都不做忽略任務

public static class DiscardPolicy implements RejectedExecutionHandler { public DiscardPolicy() { } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { }}

DiscardOldestPolicy 如果執行緒池沒有被關閉那麼丟棄佇列頭部的任務,然後提交此任務

這裡其實不一定是等待最久的任務,如果是優先順序佇列,意味著是優先順序最高的任務

public static class DiscardOldestPolicy implements RejectedExecutionHandler { public DiscardOldestPolicy() { } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e。isShutdown()) { e。getQueue()。poll(); e。execute(r); } }}

四丶Executors中的執行緒池

Executors 和 Executor的關係類似於Collection 和Collections的關係,Executors 是一個工具類,下面我們看下里面提供的一些實用執行緒池

newFixedThreadPool 固定數目工作執行緒,無界任務阻塞佇列(可以容納int最大個任務)的執行緒池——容易oom,如果請求量大容易操作阻塞佇列積壓過多工造成oom

雖然這裡keepAliveTime是0,但是執行緒池中的執行緒都是核心執行緒,核心執行緒預設不允許回收

public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit。MILLISECONDS, new LinkedBlockingQueue());}

newSingleThreadExecutor 單執行緒,無界任務阻塞佇列的執行緒池

public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit。MILLISECONDS, new LinkedBlockingQueue()));}

FinalizableDelegatedExecutorService ——執行緒池被回收的時候會呼叫shutdown方法,並且只暴露了ExecutorService介面中的方法給我們呼叫

newCachedThreadPool,支援工作執行緒數達到Integer。MAX_VALUE,且空閒時間達到60秒那麼就會被回收,使用的是SynchronousQueue不會容納任何任務,每一個任務提交之後都必須有另外一個執行緒獲取任務——執行緒多並不意味著效率高,上下文的切換,執行緒的new 和消耗都是消耗大量資源的,支援Integer。MAX_VALUE個執行緒顯然也是不符合實際的

public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer。MAX_VALUE, 60L, TimeUnit。SECONDS, new SynchronousQueue());}

newScheduledThreadPool 定時執行任務的執行緒池,還有單執行緒定時執行任務的執行緒池

使用的是ScheduledThreadPoolExecutor,後續整理這部分的原始碼

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize);}

unconfigurableExecutorService 禁止調整執行緒池引數配置的執行緒池

newWorkStealingPool 工作竊取池,使用的是ForkJoinPool,後續會學習這部分原始碼

頂部