[本文PDF地址：http://m.tkk7.com/Files/xylz/Inside.Java.Concurrency_31.ThreadPool.part4_RejectedPolicy.pdf]
上一節(jié)中提到關(guān)閉線程池過程中需要對新提交的任務(wù)進行處理。這個是java.util.concurrent.RejectedExecutionHandler處理的邏輯。
在沒有分析線程池原理之前先來分析下為什么有任務(wù)拒絕的情況發(fā)生。
這里先假設(shè)一個前提：線程池有一個任務(wù)隊列，用于緩存所有待處理的任務(wù)，正在處理的任務(wù)將從任務(wù)隊列中移除。因此在任務(wù)隊列長度有限的情況下就會出現(xiàn)新任務(wù)的拒絕處理問題，需要有一種策略來處理應(yīng)該加入任務(wù)隊列卻因為隊列已滿無法加入的情況。另外在線程池關(guān)閉的時候也需要對任務(wù)加入隊列操作進行額外的協(xié)調(diào)處理。
RejectedExecutionHandler提供了四種方式來處理任務(wù)拒絕策略。
這四種策略是獨立無關(guān)的，是對任務(wù)拒絕處理的四中表現(xiàn)形式。最簡單的方式就是直接丟棄任務(wù)。但是卻有兩種方式，到底是該丟棄哪一個任務(wù)，比如可以丟棄當前將要加入隊列的任務(wù)本身（DiscardPolicy）或者丟棄任務(wù)隊列中最舊任務(wù)（DiscardOldestPolicy）。丟棄最舊任務(wù)也不是簡單的丟棄最舊的任務(wù)，而是有一些額外的處理。除了丟棄任務(wù)還可以直接拋出一個異常（RejectedExecutionException），這是比較簡單的方式。拋出異常的方式（AbortPolicy）盡管實現(xiàn)方式比較簡單，但是由于拋出一個RuntimeException，因此會中斷調(diào)用者的處理過程。除了拋出異常以外還可以不進入線程池執(zhí)行，在這種方式（CallerRunsPolicy）中任務(wù)將有調(diào)用者線程去執(zhí)行。
上面是一些理論知識，下面結(jié)合一些例子進行分析討論。
package xylz.study.concurrency;
import java.lang.reflect.Field;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import java.util.concurrent.ThreadPoolExecutor.DiscardPolicy;
public class ExecutorServiceDemo {
static void log(String msg) {
System.out.println(System.currentTimeMillis() + " -> " + msg);
}
static int getThreadPoolRunState(ThreadPoolExecutor pool) throws Exception {
Field f = ThreadPoolExecutor.class.getDeclaredField("runState");
f.setAccessible(true);
int v = f.getInt(pool);
return v;
}
public static void main(String[] args) throws Exception {
ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(1));
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
for (int i = 0; i < 10; i++) {
final int index = i;
pool.submit(new Runnable() {
public void run() {
log("run task:" + index + " -> " + Thread.currentThread().getName());
try {
Thread.sleep(1000L);
} catch (Exception e) {
e.printStackTrace();
}
log("run over:" + index + " -> " + Thread.currentThread().getName());
}
});
}
log("before sleep");
Thread.sleep(4000L);
log("before shutdown()");
pool.shutdown();
log("after shutdown(),pool.isTerminated=" + pool.isTerminated());
pool.awaitTermination(1000L, TimeUnit.SECONDS);
log("now,pool.isTerminated=" + pool.isTerminated() + ", state="
+ getThreadPoolRunState(pool));
}
}
第一種方式直接丟棄（DiscardPolicy）的輸出結(jié)果是：
1294494050696 -> run task:0
1294494050696 -> before sleep
1294494051697 -> run over:0 -> pool-1-thread-1
1294494051697 -> run task:1
1294494052697 -> run over:1 -> pool-1-thread-1
1294494054697 -> before shutdown()
1294494054697 -> after shutdown(),pool.isTerminated=false
1294494054698 -> now,pool.isTerminated=true, state=3
對于上面的結(jié)果需要補充幾點。
- 線程池設(shè)定線程大小為1，因此輸出的線程就只有一個”pool-1-thread-1”，至于為什么是這個名稱，以后會分析。
- 任務(wù)隊列的大小為1，因此可以輸出一個任務(wù)執(zhí)行結(jié)果。但是由于線程本身可以帶有一個任務(wù)，因此實際上一共執(zhí)行了兩個任務(wù)(task0和task1)。
- shutdown()一個線程并不能理解是線程運行狀態(tài)位terminated，可能需要稍微等待一點時間。盡管這里等待時間參數(shù)是1000秒，但是實際上從輸出時間來看僅僅等了約1ms。
- 直接丟棄任務(wù)是丟棄將要進入線程池本身的任務(wù)，所以當運行task0是，task1進入任務(wù)隊列，task2~task9都被直接丟棄了，沒有運行。
如果把策略換成丟棄最舊任務(wù)（DiscardOldestPolicy），結(jié)果會稍有不同。
1294494484622 -> run task:0
1294494484622 -> before sleep
1294494485622 -> run over:0 -> pool-1-thread-1
1294494485622 -> run task:9
1294494486622 -> run over:9 -> pool-1-thread-1
1294494488622 -> before shutdown()
1294494488622 -> after shutdown(),pool.isTerminated=false
1294494488623 -> now,pool.isTerminated=true, state=3
這里依然只是執(zhí)行兩個任務(wù)，但是換成了任務(wù)task0和task9。實際上task1~task8還是進入了任務(wù)隊列，只不過被task9擠出去了。
對于異常策略（AbortPolicy）就比較簡單，這回調(diào)用線程的任務(wù)執(zhí)行。
對于調(diào)用線程執(zhí)行方式（CallerRunsPolicy），輸出的結(jié)果就有意思了。
1294496076266 -> run task:2 -> main
1294496076266 -> run task:0 -> pool-1-thread-1
1294496077266 -> run over:0 -> pool-1-thread-1
1294496077266 -> run task:1 -> pool-1-thread-1
1294496077266 -> run over:2 -> main
1294496077266 -> run task:4 -> main
1294496078267 -> run over:4 -> main
1294496078267 -> run task:5 -> main
1294496078267 -> run over:1 -> pool-1-thread-1
1294496078267 -> run task:3 -> pool-1-thread-1
1294496079267 -> run over:3 -> pool-1-thread-1
1294496079267 -> run over:5 -> main
1294496079267 -> run task:7 -> main
1294496079267 -> run task:6 -> pool-1-thread-1
1294496080267 -> run over:7 -> main
1294496080267 -> run task:9 -> main
1294496080267 -> run over:6 -> pool-1-thread-1
1294496080267 -> run task:8 -> pool-1-thread-1
1294496081268 -> run over:9 -> main
1294496081268 -> before sleep
1294496081268 -> run over:8 -> pool-1-thread-1
1294496085268 -> before shutdown()
1294496085268 -> after shutdown(),pool.isTerminated=false
1294496085269 -> now,pool.isTerminated=true, state=3
由于啟動線程有稍微的延時，因此一種可能的執(zhí)行順序是這樣的。
- 首先pool-1-thread-1線程執(zhí)行task0,同時將task1加入任務(wù)隊列（submit(task1)）。
- 對于task2，由于任務(wù)隊列已經(jīng)滿了，因此有調(diào)用線程main執(zhí)行（execute(task2)）。
- 在mian等待task2任務(wù)執(zhí)行完畢，對于任務(wù)task3，由于此時任務(wù)隊列已經(jīng)空了，因此task3將進入任務(wù)隊列。
- 此時main線程是空閑的，因此對于task4將由main線程執(zhí)行。此時pool-1-thread-1線程可能在執(zhí)行任務(wù)task1。任務(wù)隊列中依然有任務(wù)task3。
- 因此main線程執(zhí)行完畢task4后就立即執(zhí)行task5。
- 很顯然task1執(zhí)行完畢，task3被線程池執(zhí)行，因此task6進入任務(wù)隊列。此時task7被main線程執(zhí)行。
- task6開始執(zhí)行時，task8進入任務(wù)隊列。main線程開始執(zhí)行task9。
- 然后線程池執(zhí)行線程task8結(jié)束。
- 整個任務(wù)隊列執(zhí)行完畢，線程池完畢。
如果有興趣可以看看ThreadPoolExecutor中四種RejectedExecutionHandler的源碼，都非常簡單。
©2009-2014 IMXYLZ
|求賢若渴