-
之前我們介紹了線程池的四種拒絕策略,了解了線程池參數的含義,那么今天我們來聊聊
Java
中常見的幾種線程池,以及在jdk7
加入的ForkJoin
新型線程池 -
首先我們列出
Java
中的六種線程池如下
線程池名稱 | 描述 |
---|---|
FixedThreadPool | 核心線程數與最大線程數相同 |
SingleThreadExecutor | 一個線程的線程池 |
CachedThreadPool | 核心線程為0,最大線程數為Integer. MAX_VALUE |
ScheduledThreadPool | 指定核心線程數的定時線程池 |
SingleThreadScheduledExecutor | 單例的定時線程池 |
ForkJoinPool | JDK 7 新加入的一種線程池 |
- 在了解集中線程池時我們先來熟悉一下主要幾個類的關系,
ThreadPoolExecutor
的類圖,以及Executors
的主要方法:
- 上面看到的類圖,方便幫助下面的理解和查看,我們可以看到一個核心類
ExecutorService
, 這是我們線程池都實現的基類,我們接下來說的都是它的實現類。
FixedThreadPool
FixedThreadPool
線程池的特點是它的核心線程數和最大線程數一樣,我們可以看它的實現代碼在Executors#newFixedThreadPool(int)
中,如下:
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
我們可以看到方法內創建線程調用的實際是
ThreadPoolExecutor
類,這是線程池的核心執行器,傳入的nThread
參數作為核心線程數和最大線程數傳入,隊列采用了一個鏈表結構的有界隊列。
- 這種線程池我們可以看作是固定線程數的線程池,它只有在開始初始化的時候線程數會從0開始創建,但是創建好后就不再銷毀,而是全部作為常駐線程池,這里如果對線程池參數不理解的可以看之前文章 《解釋線程池各個參數的含義》。
- 對于這種線程池他的第三個和第四個參數是沒意義,它們是空閑線程存活時間,這里都是常駐不存在銷毀,當線程處理不了時會加入到阻塞隊列,這是一個鏈表結構的有界阻塞隊列,最大長度是Integer. MAX_VALUE
SingleThreadExecutor
SingleThreadExecutor
線程的特點是它的核心線程數和最大線程數均為1,我們也可以將其任務是一個單例線程池,它的實現代碼是Executors#newSingleThreadExcutor()
, 如下:
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory));
}
- 上述代碼中我們發現它有一個重載函數,傳入了一個
ThreadFactory
的參數,一般在我們開發中會傳入我們自定義的線程創建工廠,如果不傳入則會調用默認的線程工廠 - 我們可以看到它與
FixedThreadPool
線程池的區別僅僅是核心線程數和最大線程數改為了1,也就是說不管任務多少,它只會有唯一的一個線程去執行 - 如果在執行過程中發生異常等導致線程銷毀,線程池也會重新創建一個線程來執行后續的任務
- 這種線程池非常適合所有任務都需要按被提交的順序來執行的場景,是個單線程的串行。
CachedThreadPool
cachedThreadPool
線程池的特點是它的常駐核心線程數為0,正如其名字一樣,它所有的縣城都是臨時的創建,關于它的實現在Executors#newCachedThreadPool()
中,代碼如下:
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
threadFactory);
}
- 從上述代碼中我們可以看到
CachedThreadPool
線程池中,最大線程數為Integer.MAX_VALUE
, 意味著他的線程數幾乎可以無限增加。 - 因為創建的線程都是臨時線程,所以他們都會被銷毀,這里空閑 線程銷毀時間是60秒,也就是說當線程在60秒內沒有任務執行則銷毀
- 這里我們需要注意點,它使用了
SynchronousQueue
的一個阻塞隊列來存儲任務,這個隊列是無法存儲的,因為他的容量為0,它只負責對任務的傳遞和中轉,效率會更高,因為核心線程都為0,這個隊列如果存儲任務不存在意義。
ScheduledThreadPool
ScheduledThreadPool
線程池是支持定時或者周期性執行任務,他的創建代碼Executors.newSchedsuledThreadPool(int)
中,如下所示:
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
public static ScheduledExecutorService newScheduledThreadPool(
int corePoolSize, ThreadFactory threadFactory) {
return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}
- 我們發現這里調用了
ScheduledThreadPoolExecutor
這個類的構造函數,進一步查看發現ScheduledThreadPoolExecutor
類是一個繼承了ThreadPoolExecutor
的,同時實現了ScheduledExecutorService
接口,我們看到它的幾個構造函數都是調用父類ThreadPoolExecutor
的構造函數
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory);
}
public ScheduledThreadPoolExecutor(int corePoolSize,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), handler);
}
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory, handler);
}
- 從上面代碼我們可以看到和其他線程池創建并沒有差異,只是這里的任務隊列是
DelayedWorkQueue
關于阻塞丟列我們下篇文章專門說,這里我們先創建一個周期性的線程池來看一下
public static void main(String[] args) {
ScheduledExecutorService service = Executors.newScheduledThreadPool(5);
// 1. 延遲一定時間執行一次
service.schedule(() ->{
System.out.println("schedule ==> 云棲簡碼-i-code.online");
},2, TimeUnit.SECONDS);
// 2. 按照固定頻率周期執行
service.scheduleAtFixedRate(() ->{
System.out.println("scheduleAtFixedRate ==> 云棲簡碼-i-code.online");
},2,3,TimeUnit.SECONDS);
//3. 按照固定頻率周期執行
service.scheduleWithFixedDelay(() -> {
System.out.println("scheduleWithFixedDelay ==> 云棲簡碼-i-code.online");
},2,5,TimeUnit.SECONDS);
}
- 上面代碼是我們簡單創建了
newScheduledThreadPool
,同時演示了里面的三個核心方法,首先看執行的結果:
- 首先我們看第一個方法
schedule
, 它有三個參數,第一個參數是線程任務,第二個delay
表示任務執行延遲時長,第三個unit
表示延遲時間的單位,如上面代碼所示就是延遲兩秒后執行任務
public ScheduledFuture<?> schedule(Runnable command,
long delay, TimeUnit unit);
- 第二個方法是
scheduleAtFixedRate
如下, 它有四個參數,command
參數表示執行的線程任務 ,initialDelay
參數表示第一次執行的延遲時間,period
參數表示第一次執行之后按照多久一次的頻率來執行,最后一個參數是時間單位。如上面案例代碼所示,表示兩秒后執行第一次,之后按每隔三秒執行一次
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
- 第三個方法是
scheduleWithFixedDelay
如下,它與上面方法是非常類似的,也是周期性定時執行, 參數含義和上面方法一致。這個方法和scheduleAtFixedRate
的區別主要在于時間的起點計時不同
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
scheduleAtFixedRate
是以任務開始的時間為時間起點來計時,時間到就執行第二次任務,與任務執行所花費的時間無關;而scheduleWithFixedDelay
是以任務執行結束的時間點作為計時的開始。如下所示
SingleThreadScheduledExecutor
- 它實際和
ScheduledThreadPool
線程池非常相似,它只是ScheduledThreadPool
的一個特例,內部只有一個線程,它只是將ScheduledThreadPool
的核心線程數設置為了 1。如源碼所示:
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
}
- 上面我們介紹了五種常見的線程池,對于這些線程池我們可以從核心線程數、最大線程數、存活時間三個維度進行一個簡單的對比,有利于我們加深對這幾種線程池的記憶。
FixedThreadPool | SingleThreadExecutor | CachedThreadPool | ScheduledThreadPool | SingleThreadScheduledExecutor | |
---|---|---|---|---|---|
corePoolSize | 構造函數傳入 | 1 | 0 | 構造函數傳入 | 1 |
maxPoolSize | 同corePoolSize | 1 | Integer. MAX_VALUE | Integer. MAX_VALUE | Integer. MAX_VALUE |
keepAliveTime | 0 | 0 | 60 | 0 | 0 |
ForkJoinPool
ForkJoinPool
這是一個在JDK7
引入的新新線程池,它的主要特點是可以充分利用多核CPU
, 可以把一個任務拆分為多個子任務,這些子任務放在不同的處理器上并行執行,當這些子任務執行結束后再把這些結果合并起來,這是一種分治思想。ForkJoinPool
也正如它的名字一樣,第一步進行Fork
拆分,第二步進行Join
合并,我們先來看一下它的類圖結構
ForkJoinPool
的使用也是通過調用submit(ForkJoinTask<T> task)
或invoke(ForkJoinTask<T> task)
方法來執行指定任務了。其中任務的類型是ForkJoinTask
類,它代表的是一個可以合并的子任務,他本身是一個抽象類,同時還有兩個常用的抽象子類RecursiveAction
和RecursiveTask
,其中RecursiveTask
表示的是有返回值類型的任務,而RecursiveAction
則表示無返回值的任務。下面是它們的類圖:
- 下面我們通過一個簡單的代碼先來看一下如何使用
ForkJoinPool
線程池
/**
* @url: i-code.online
* @author: AnonyStar
* @time: 2020/11/2 10:01
*/
public class ForkJoinApp1 {
/**
目標: 打印0-200以內的數字,進行分段每個間隔為10以上,測試forkjoin
*/
public static void main(String[] args) {
// 創建線程池,
ForkJoinPool joinPool = new ForkJoinPool();
// 創建根任務
SubTask subTask = new SubTask(0,200);
// 提交任務
joinPool.submit(subTask);
//讓線程阻塞等待所有任務完成 在進行關閉
try {
joinPool.awaitTermination(2, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
joinPool.shutdown();
}
}
class SubTask extends RecursiveAction {
int startNum;
int endNum;
public SubTask(int startNum,int endNum){
super();
this.startNum = startNum;
this.endNum = endNum;
}
@Override
protected void compute() {
if (endNum - startNum < 10){
// 如果分裂的兩者差值小于10 則不再繼續,直接打印
System.out.println(Thread.currentThread().getName()+": [startNum:"+startNum+",endNum:"+endNum+"]");
}else {
// 取中間值
int middle = (startNum + endNum) / 2;
//創建兩個子任務,以遞歸思想,
SubTask subTask = new SubTask(startNum,middle);
SubTask subTask1 = new SubTask(middle,endNum);
//執行任務, fork() 表示異步的開始執行
subTask.fork();
subTask1.fork();
}
}
}
結果:
- 從上面的案例我們可以看到我們,創建了很多個線程執行,因為我測試的電腦是12線程的,所以這里實際是創建了12個線程,也側面說明了充分調用了每個處理的線程處理能力
- 上面案例其實我們發現很熟悉的味道,那就是以前接觸過的遞歸思想,將上面的案例圖像化如下,更直觀的看到,
- 上面的例子是無返回值的案例,下面我們來看一個典型的有返回值的案例,相信大家都聽過及很熟悉斐波那契數列,這個數列有個特點就是最后一項的結果等于前兩項的和,如:
0,1,1,2,3,5...f(n-2)+f(n-1)
, 即第0項為0 ,第一項為1,則第二項為0+1=1
,以此類推。我們最初的解決方法就是使用遞歸來解決,如下計算第n項的數值:
private int num(int num){
if (num <= 1){
return num;
}
num = num(num-1) + num(num -2);
return num;
}
- 從上面簡單代碼中可以看到,當
n<=1
時返回n
, 如果n>1
則計算前一項的值f1
,在計算前兩項的值f2
, 再將兩者相加得到結果,這就是典型的遞歸問題,也是對應我們的ForkJoin
的工作模式,如下所示,根節點產生子任務,子任務再次衍生出子子任務,到最后在進行整合匯聚,得到結果。
- 我們通過
ForkJoinPool
來實現斐波那契數列的計算,如下展示:
/**
* @url: i-code.online
* @author: AnonyStar
* @time: 2020/11/2 10:01
*/
public class ForkJoinApp3 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ForkJoinPool pool = new ForkJoinPool();
//計算第二是項的數值
final ForkJoinTask<Integer> submit = pool.submit(new Fibonacci(20));
// 獲取結果,這里獲取的就是異步任務的最終結果
System.out.println(submit.get());
}
}
class Fibonacci extends RecursiveTask<Integer>{
int num;
public Fibonacci(int num){
this.num = num;
}
@Override
protected Integer compute() {
if (num <= 1) return num;
//創建子任務
Fibonacci subTask1 = new Fibonacci(num - 1);
Fibonacci subTask2 = new Fibonacci(num - 2);
// 執行子任務
subTask1.fork();
subTask2.fork();
//獲取前兩項的結果來計算和
return subTask1.join()+subTask2.join();
}
}
-
通過
ForkJoinPool
可以極大的發揮多核處理器的優勢,尤其非常適合用于遞歸的場景,例如樹的遍歷、最優路徑搜索等場景。 -
上面說的是
ForkJoinPool
的使用上的,下面我們來說一下其內部的構造,對于我們前面說的幾種線程池來說,它們都是里面只有一個隊列,所有的線程共享一個。但是在ForkJoinPool
中,其內部有一個共享的任務隊列,除此之外每個線程都有一個對應的雙端隊列Deque
, 當一個線程中任務被Fork
分裂了,那么分裂出來的子任務就會放入到對應的線程自己的Deque
中,而不是放入公共隊列。這樣對于每個線程來說成本會降低很多,可以直接從自己線程的隊列中獲取任務而不需要去公共隊列中爭奪,有效的減少了線程間的資源競爭和切換。
- 有一種情況,當線程有多個如
t1,t2,t3...
,在某一段時間線程t1
的任務特別繁重,分裂了數十個子任務,但是線程t0
此時卻無事可做,它自己的deque
隊列為空,這時為了提高效率,t0
就會想辦法幫助t1
執行任務,這就是“work-stealing
”的含義。 - 雙端隊列
deque
中,線程t1
獲取任務的邏輯是后進先出,也就是LIFO(Last In Frist Out)
,而線程t0
在“steal
”偷線程t1
的deque
中的任務的邏輯是先進先出,也就是FIFO(Fast In Frist Out)
,如圖所示,圖中很好的描述了兩個線程使用雙端隊列分別獲取任務的情景。你可以看到,使用 “work-stealing
” 算法和雙端隊列很好地平衡了各線程的負載。
本文由AnonyStar 發布,可轉載但需聲明原文出處。
歡迎關注微信公賬號 :云棲簡碼 獲取更多優質文章
更多文章關注筆者博客 :云棲簡碼 i-code.online