背景

面試的時候經常會被三連問。用過嗎？如何用的？場景是什么？所以有必要好好的研究下線程池迫在眉睫。

1、講解之前先了解下 retry: 因為源碼中有這個retry標記

先看一個簡單的例子

/*** @author shuliangzhao* @Title: RetryTest* @ProjectName design-parent* @Description: TODO* @date 2019/6/1 23:43*/ public class RetryTest {public static void main(String[] args) {testRetry();}public static void testRetry() {//retry:注釋1for (int i = 0; i < 10; i++) {retry: //注釋2while (i == 5) {continue retry;}System.out.print(i + " ");}} }

如上如果只保留注釋1，循環到 i==5的時候，程序跳到retry的那一行開始執行，此時 i 的值未變，然后又是i==5，程序進入死循環一直執行4到6行；執行結果為0 1 2 3 4

如果直流注釋2，循環到 i==5的時候，程序跳到retry的那一行開始執行，注意此時 i 的值還是5，接著 i++（i 不是從0開始了），所以輸出 0 1 2 3 4 6 7 8 9

說明：其實retry就是一個標記，標記程序跳出循環的時候從哪里開始執行，功能類似于goto。retry一般都是跟隨者for循環出現，第一個retry的下面一行就是for循環，而且第二個retry的前面一般是 continue或是 break。

2、為什么要使用線程池

缺點
a、每次new Thread新建對象，性能差。
b、缺乏統一管理，可能無限制的新建線程，過多占用系統資源導致死機或OOM
優點
a、重用存在的線程，減少對象創建，消亡的開銷
b、有效控制最大并發線程數，提高系統資源利用率

3、線程池實現原理

當線程提交一個任務時候，如果處理請看下圖

?

image.png

ThreadPoolExecutor執行execute()分4種情況

a、若當前運行的線程少于corePoolSize,則創建新線程來執行任務(執行這一步需要獲取全局鎖)
b、若運行的線程多于或等于corePoolSize,則將任務加入BlockingQueue
c、若無法將任務加入BlockingQueue,則創建新的線程來處理任務(執行這一步需要獲取全局鎖)
d、若創建新線程將使當前運行的線程超出maximumPoolSize,任務將被拒絕,并調用RejectedExecutionHandler.rejectedExecution()
采取上述思路,是為了在執行execute()時,盡可能避免獲取全局鎖
在ThreadPoolExecutor完成預熱之后（當前運行的線程數大于等于corePoolSize),幾乎所有的execute()方法調用都是執行步驟b,而步驟b不需要獲取全局鎖
源碼分析execute()

/*** Executes the given task sometime in the future. The task* may execute in a new thread or in an existing pooled thread.** If the task cannot be submitted for execution, either because this* executor has been shutdown or because its capacity has been reached,* the task is handled by the current {@code RejectedExecutionHandler}.** @param command the task to execute* @throws RejectedExecutionException at discretion of* {@code RejectedExecutionHandler}, if the task* cannot be accepted for execution* @throws NullPointerException if {@code command} is null*/public void execute(Runnable command) {if (command == null)throw new NullPointerException();/** Proceed in 3 steps:** 1. If fewer than corePoolSize threads are running, try to* start a new thread with the given command as its first* task. The call to addWorker atomically checks runState and* workerCount, and so prevents false alarms that would add* threads when it shouldn't, by returning false.** 2. If a task can be successfully queued, then we still need* to double-check whether we should have added a thread* (because existing ones died since last checking) or that* the pool shut down since entry into this method. So we* recheck state and if necessary roll back the enqueuing if* stopped, or start a new thread if there are none.** 3. If we cannot queue task, then we try to add a new* thread. If it fails, we know we are shut down or saturated* and so reject the task.*///表示 “線程池狀態” 和 “線程數” 的整數int c = ctl.get();// 如果當前線程數少于核心線程數，直接添加一個 worker 執行任務，// 創建一個新的線程，并把當前任務 command 作為這個線程的第一個任務(firstTask)if (workerCountOf(c) < corePoolSize) {// 添加任務成功，即結束// 執行的結果，會包裝到 FutureTask // 返回 false 代表線程池不允許提交任務if (addWorker(command, true))return;c = ctl.get();}// 到這說明，要么當前線程數大于等于核心線程數，要么剛剛 addWorker 失敗// 如果線程池處于 RUNNING ，把這個任務添加到任務隊列 workQueue 中if (isRunning(c) && workQueue.offer(command)) {/* 若任務進入 workQueue，我們是否需要開啟新的線程* 線程數在 [0, corePoolSize) 是無條件開啟新線程的* 若線程數已經大于等于 corePoolSize，則將任務添加到隊列中，然后進到這里*/int recheck = ctl.get();// 若線程池不處于 RUNNING ，則移除已經入隊的這個任務，并且執行拒絕策略if (! isRunning(recheck) && remove(command))reject(command);// 若線程池還是 RUNNING ，且線程數為 0，則開啟新的線程// 這塊代碼的真正意圖：擔心任務提交到隊列中了，但是線程都關閉了else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 若 workQueue 滿，到該分支// 以 maximumPoolSize 為界創建新 worker，// 若失敗，說明當前線程數已經達到 maximumPoolSize，執行拒絕策略else if (!addWorker(command, false))reject(command);}

其他源碼暫不貼出來了，自己可以認真閱讀下。

4、線程池創建

我們可以通過ThreadPoolExecutor來創建一個線程池
創建一個線程池時需要的參數
corePoolSize(核心線程數量)
線程池中應該保持的主要線程的數量.即使線程處于空閑狀態，除非設置了allowCoreThreadTimeOut這個參數,當提交一個任務到線程池時,若線程數量<corePoolSize,線程池會創建一個新線程放入works(一個HashSet)中執行任務,即使其他空閑的基本線程能夠執行新任務也還是會創建新線程,等到需要執行的任務數大于線程池基本大小時就不再創建,會嘗試放入等待隊列workQueue(一個BlockingQueue),如果調用了線程池的prestartAllCoreThreads(),線程池會提前創建并啟動所有核心線程

workQueue
存儲待執行任務的阻塞隊列，這些任務必須是Runnable的對象（如果是Callable對象，會在submit內部轉換為Runnable對象）
runnableTaskQueue(任務隊列):用于保存等待執行的任務的阻塞隊列.可以選擇以下幾個阻塞隊列.
LinkedBlockingQueue:一個基于鏈表結構的阻塞隊列,此隊列按FIFO排序元素,吞吐量通常要高于ArrayBlockingQueue.靜態工廠方法Executors.newFixedThreadPool()使用了這個隊列
SynchronousQueue:一個不存儲元素的阻塞隊列.每個插入操作必須等到另一個線程調用移除操作,否則插入操作一直處于阻塞狀態,吞吐量通常要高于Linked-BlockingQueue,靜態工廠方法Executors.newCachedThreadPool使用了這個隊列

maximumPoolSize（線程池最大線程數）
線程池允許創建的最大線程數
若隊列滿,并且已創建的線程數小于最大線程數,則線程池會再創建新的線程放入works中執行任務,CashedThreadPool的關鍵,固定線程數的線程池無效
若使用了無界任務隊列,這個參數就沒什么效果

ThreadFactory:用于設置創建線程的工廠,可以通過線程工廠給每個創建出來的線程設置更有意義的名字.使用開源框架guava提供ThreadFactoryBuilder可以快速給線程池里的線程設置有意義的名字,代碼如下
new ThreadFactoryBuilder().setNameFormat("XX-task-%d").build();

RejectedExecutionHandler（飽和策略）:當隊列和線程池都滿,說明線程池處于飽和,必須采取一種策略處理提交的新任務.策略默認AbortPolicy,表無法處理新任務時拋出異常.在JDK 1.5中Java線程池框架提供了以下4種策略
AbortPolicy：丟棄任務，拋出 RejectedExecutionException
CallerRunsPolicy:只用調用者所在線程來運行任務,有反饋機制，使任務提交的速度變慢）。
DiscardOldestPolicy
若沒有發生shutdown,嘗試丟棄隊列里最近的一個任務,并執行當前任務, 丟棄任務緩存隊列中最老的任務，并且嘗試重新提交新的任務
DiscardPolicy:不處理,丟棄掉, 拒絕執行，不拋異常
當然,也可以根據應用場景需要來實現RejectedExecutionHandler接口自定義策略.如記錄日志或持久化存儲不能處理的任務

keepAliveTime（線程活動保持時間）
線程沒有任務執行時最多保持多久時間終止
線程池的工作線程空閑后，保持存活的時間。
所以，如果任務很多，并且每個任務執行的時間比較短，可以調大時間，提高線程的利用率

TimeUnit（線程活動保持時間的單位）:指示第三個參數的時間單位；可選的單位有天（DAYS）、小時（HOURS）、分鐘（MINUTES）、毫秒（MILLISECONDS）、微秒（MICROSECONDS，千分之一毫秒）和納秒（NANOSECONDS，千分之一微秒）
可以使用Executors創建線程池

?

image.png

?

使用線程池例子

/*** @author shuliangzhao* @Title: ThreadTaskId* @ProjectName design-parent* @Description: TODO* @date 2019/6/1 23:03*/ public class ThreadTaskId implements Runnable {private final int id;public ThreadTaskId(int id) {this.id = id;}@Overridepublic void run() {for (int i = 0;i < 5;i++) {System.out.println("TaskInPool-["+id+"] is running phase-"+i);try {TimeUnit.SECONDS.sleep(1);System.out.println("TaskInPool-["+id+"] is over");} catch (InterruptedException e) {e.printStackTrace();}}} }

客戶端

/*** @author shuliangzhao* @Title: ThreadPoolExample* @ProjectName design-parent* @Description: TODO* @date 2019/6/1 23:03*/ public class ThreadPoolExample {public static void main(String[] args) {ExecutorService executorService = Executors.newSingleThreadExecutor();for (int i = 0; i < 5; i++) {executorService.execute(new ThreadTaskId(i));}executorService.shutdown();} }

執行結果

image.png

以上就是線程池的簡單介紹，這個不是完善版本，會繼續補充的。_

總結

以上是生活随笔為你收集整理的Java线程池(Executor)详解和用法的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。