執行緒池底層原理你真的懂嗎？

與執行緒池相關的介面實作類關系
Exectutor是介面 Executors 是輔助工具類 就好像是Collection 介面 Collections 工具類

使用場景：
對比new Thread() 創建和銷毀執行緒都非常耗時，使用執行緒池可以達到執行緒復用/重用，

池化：享元模式，如連接池，常量池

執行緒池的底層原理及七大引數的意義
七大引數
public ThreadPoolExecutor（
int corePoolSize,//核心執行緒數
int maximumPoolSize,//最大執行緒數
long keepAliveTime,
TimeUnit unit, //臨時執行緒空閑時間
ThreadFactory threadFactory ,//執行緒工廠
BlockingQueue<Runnable> workQueue,//阻塞佇列
RejectedExecutionHandler handler //拒絕策略
)

引數 作用
corePoolSize 核心執行緒數（銀行今日當值網視窗）
maximumPoolSize 最大執行緒數（銀行最多的視窗也就是所有的，最多容納的，物理視窗上限）
keepAliveTime 空閑執行緒最大存活時間 當前執行緒池數量超過corePoolSize并且時間達到keepAliveTime，多余的空閑執行緒就會被銷毀直到執行緒的數量只剩下corePoolSize
TimeUnitkeepAliveTime 時間單位
workQueue 阻塞任務佇列（corePoolSize用完了，客戶先去銀行里面的候客區）
threadFactory 新建執行緒工廠（默認，銀行網點的logo/作業人員的制服等等都是默認的，都是標配的）
RejectedExecutionHandler 拒絕策略 當提交任務數超過 maxmumPoolSize+workQueue 之和時（所有視窗都用上了并且候客區也滿了），任務會交給RejectedExecutionHandler 來處理
3-4 空閑執行緒的存活時間 （加班視窗如果沒有業務量的最多的等待時間），沒有業務那么執行緒數量會慢慢回退，直到執行緒的數量只剩下corePoolSize

作業原理 很重要！

1）如果當前運行的執行緒少于corePoolSize，則創建新執行緒來執行任務（注意，執行這一步驟需要
獲取全域鎖），
2）如果運行的執行緒等于或多于corePoolSize，則將任務加入BlockingQueue，
3）如果無法將任務加入BlockingQueue（佇列已滿），則創建新的執行緒來處理任務（注意，執行
這一步驟需要獲取全域鎖），
4）如果創建新執行緒將使當前運行的執行緒超出maximumPoolSize，任務將被拒絕，并呼叫RejectedExecutionHandler.rejectedExecution()方法

執行緒池中的執行緒執行任務分兩種情況，如下，

在execute()方法中創建一個執行緒時，會讓這個執行緒執行當前任務，
這個執行緒執行完當前任務后，會反復從BlockingQueue獲取任務來執行，
特科
執行緒池的拒絕策略你談談？
是什么

等待佇列已經滿了，再也塞不下新的任務，同時執行緒池中的執行緒數達到了最大執行緒數，無法繼續為新任務服務，
均實作了RejectedExecutionHandler介面

拒絕策略
jdk內置的拒絕策略

AbortPolicy：直接拋出例外，(默認采用此策略)，處理程式遭到拒絕將拋出運行時 RejectedExecutionException
CallerRunsPolicy：只用呼叫者所在執行緒來運行任務，執行緒呼叫運行該任務的 execute 本身，此策略提供簡單的反饋控制機制，能夠級訓新任務的提交速度，也就是說如果執行緒池執行不了任務了，就由main執行緒（main執行緒呼叫的執行緒池）來執行執行緒池執行不了的任務
DiscardPolicy：不處理，丟棄掉，，不能執行的任務將被洗掉
DiscardOldestPolicy：：丟棄佇列里最近的一個任務，并執行當前任務，，如果執行程式尚未關閉，則位于作業佇列頭部的任務將被洗掉，然后重試執行程式（如果再次失敗，則重復此程序）
自定義執行緒池的使用
execute()方法用于提交不需要回傳值的任務，所以無法判斷任務是否被執行緒池執行成功，

submit 不僅可以用于多執行緒的提交還有Callable，是帶回傳值的

public static void main(String[] args) {

//手寫引數
ExecutorService threadPool = new ThreadPoolExecutor(
2,//核心
5,//最大
1L, //空閑時間 單位是long
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(3),//任務佇列
Executors.defaultThreadFactory(),//執行緒工廠
//直接拋出例外
new ThreadPoolExecutor.AbortPolicy()

//main執行緒呼叫的是這個執行緒 誰呼叫我我回退給誰
//你從哪個銀行來我們銀行辦理業務的，你就回哪去
/*new ThreadPoolExecutor.CallerRunsPolicy()*/

//這下面兩個都最多執行的maxmunPoolSize 也就是8 剩下的直接拋棄
/*new ThreadPoolExecutor.DiscardOldestPolicy()*/
/* new ThreadPoolExecutor.DiscardPolicy()*/
);
try {
//模擬十個執行緒變數業務
for (int i = 0; i < 9; i++) {
threadPool.execute(()->{
System.out.println(Thread.currentThread().getName());
});
}
}catch (Exception e) {
e.printStackTrace();
}finally {
threadPool.shutdown();//關閉執行緒池
}
}

submit()方法用于提交需要回傳值的任務，執行緒池會回傳一個future型別的物件，通過這個future物件可以判斷任務是否執行成功，并且可以通過future的get()方法來獲取回傳值，get()方法會阻塞當前執行緒直到任務完成，也就是雖然你使用了拒絕策略，但是在提交任務這里，是由main執行緒來做的，所以main執行緒如果獲取不到回傳值，就會阻塞不會再繼續提交新的任務，

public static void main(String[] args) {
sumbit();
}
private static void sumbit() {
//手寫引數
ExecutorService threadPool = new ThreadPoolExecutor(
2,
5,
1L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(3),
Executors.defaultThreadFactory(),
//直接拋出例外
new ThreadPoolExecutor.AbortPolicy()

//main執行緒呼叫的是這個執行緒 誰呼叫我我回退給誰
//你從哪個銀行來我們銀行辦理業務的，你就回哪去
/*new ThreadPoolExecutor.CallerRunsPolicy()*/

//這下面兩個都最多執行的maxmunPoolSize 也就是8 剩下的直接拋棄
/*new ThreadPoolExecutor.DiscardOldestPolicy()*/
/* new ThreadPoolExecutor.DiscardPolicy()*/
);
try {
//模擬十個執行緒變數業務
for (int i = 0; i < 9; i++) {
//獲取回傳值
Future<String> submit = threadPool.submit(new CallableThread());
//get()方法會阻塞當前執行緒直到任務完成
//也就是說會阻塞main執行緒 不會去繼續提交新的任務
String str = submit.get();
System.out.println(str);
}
}catch (Exception e) {
e.printStackTrace();
}finally {
threadPool.shutdown();//關閉執行緒池
}
}

執行緒池的關閉
它們的原理是遍歷執行緒池中的工作執行緒，然后逐個呼叫執行緒的interrupt方法來中斷執行緒，所以無法回應中斷的任務可能永遠無法終止

shutdown 只是中斷任務佇列里面的執行緒，不中斷正在執行的

shutdownnow 中斷任務佇列和正在運行的

JDK自帶的執行緒池
Executor框架最核心的類是ThreadPoolExecutor，它是執行緒池的實作類，通過Executor框架的工具類
Executors，可以創建3種型別的ThreadPoolExecutor，

創建固定大小的執行緒池：public static ExecutorService newFixedThreadPool(int
nThreads)
單執行緒池：public static ExecutorService newSingleThreadExecutor()
創建無大小限制的執行緒池：public static ExecutorService newCachedThreadPool()
或者定期執行任務：public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)

需要了解同步佇列的相關知識同步佇列前兩個是linkedList，后面的是同步佇列

對Java中的阻塞佇列有所了解的話，看到這里或許就能夠明白原因了，
Java中的BlockingQueue主要有兩種實作，分別是ArrayBlockingQueue 和 LinkedBlockingQueue，
ArrayBlockingQueue是一個用陣列實作的有界阻塞佇列，必須設定容量，
LinkedBlockingQueue是一個用鏈表實作的有界阻塞佇列，容量可以選擇進行設定，不設定的話，將是一個無邊界的阻塞佇列，最大長度為Integer.MAX_VALUE，

這里的問題就出在：不設定的話，將是一個無邊界的阻塞佇列，最大長度為Integer.MAX_VALUE，也就是說，如果我們不設定LinkedBlockingQueue的容量的話，其默認容量將會是Integer.MAX_VALUE，

而newFixedThreadPool中創建LinkedBlockingQueue時，并未指定容量，此時，LinkedBlockingQueue就是一個無邊界佇列，對于一個無邊界佇列來說，是可以不斷的向佇列中加入任務的，這種情況下就有可能因為任務過多而導致記憶體溢位問題，

上面提到的問題主要體現在newFixedThreadPool和newSingleThreadExecutor兩個工廠方法上，并不是說newCachedThreadPool和newScheduledThreadPool這兩個方法就安全了，這兩種方式創建的最大執行緒數可能是Integer.MAX_VALUE，而創建這么多執行緒，必然就有可能導致OOM，

為什么不用這些JDK自帶的執行緒池
執行緒池不允許使用Executors去創建，而是通過ThreadPoolExecutor的方式，阿里的編程規范中明確指出，

合理配置執行緒池你是否考慮過？
首先你需要知道cpu核心數

System.out.println(Runtime.getRuntime().availableProcessors());
1
然后根據不同的業務采用不同的策略

CPU 密集型
CPU 密集的意思是該任務需要大量的運算，而沒有阻塞，CPU 一直全速運行，
CPU 密集型任務盡可能的少的執行緒數量，一般為 CPU 核數 + 1 個執行緒的執行緒池，
IO 密集型
由于 IO 密集型任務執行緒并不是一直在執行任務，可以多分配一點執行緒數，如 CPU * 2 ，
也可以使用公式：CPU 核數 / (1 - 阻塞系數)；其中阻塞系數在 0.8 ～ 0.9 之間，

本文首發于java黑洞網，博客園同步更新

標籤：其他

上一篇：python3合并兩個有序串列 and 加一

下一篇：MySQL視圖了解一下