執行緒池之Executor框架
Java的執行緒既是作業單元，也是執行機制，從JDK5開始，把作業機單元和執行機制分離開來，作業單元包括Runnable和Callable，而執行機制由Executor框架提供，
1. Executor框架簡介
1.1 Executor框架的兩級調度模型
在上層，Java多執行緒程式通常把應用分解為若干個任務，然后使用用戶級的調度器（Executor框架）將這些任務映射為固定數量的執行緒，
在底層，作業系統內核將這些執行緒映射到硬體處理器上，
1.2 Executor框架的結構
Executor框架主要由3部分組成：
- 任務，包括被執行任務需要實作的介面：Runnable介面或者Callable介面，
- 任務的執行，包括任務執行機制的核心介面Executor，以及繼承自Executor的ExecutorService介面，Executor框架有兩個關鍵類實作了ExecutorService介面（ThreadPoolExecutor和ScheduledThreadPoolExecutor），
- 異步計算的結果，包括Future和實作Future的FutureTask類，
Executor框架的成員及其關系可以用一下的關系圖表示：
Executor框架的使用示意圖：
使用步驟：
- 主執行緒首先創建實作Runnable或Callable介面的任務物件，工具類Executors可以把一個Runnable物件封裝為一個Callable物件（
Executors.callable(Runnable task)或Executors.callable(Runnable task, Object result)）， - 創建Executor介面的實作類ThreadPoolExecutor類或者ScheduledThreadPoolExecutor類的物件，然后呼叫其execute()方法或者submit()方法把作業任務添加到執行緒中，如果有回傳值則回傳Future物件，其中Callable物件有回傳值，因此使用submit()方法；而Runnable可以使用execute()方法，此外還可以使用submit()方法，只要使用callable(Runnable task)或者callable(Runnable task, Object result)方法把Runnable物件包裝起來就可以，使用callable(Runnable task)方法回傳的null，使用callable(Runnable task, Object result)方法回傳result，
- 主執行緒可以執行Future物件的get()方法獲取回傳值，也可以呼叫cancle()方法取消當前執行緒的執行，
1.3 Executor框架的使用案例
import java.util.concurrent.*;
public class ExecutorDemo {
// 創建ThreadPoolExecutor實作類
private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
5,
10,
100,
TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(5),
);
public static void main(String[] args) {
// 采用submit()方法提交Callable物件并回傳Future物件
Future<String> future = executor.submit(new callableDemo());
try {
// get()方法獲取回傳值
System.out.println(future.get());
} catch (InterruptedException | ExecutionException e) {
// 處理例外
e.printStackTrace();
} finally {
// 關閉執行緒池
executor.shutdown();
}
}
}
/**
* 創建Callable介面的實作類
*/
class callableDemo implements Callable<String> {
@Override
public String call() throws Exception {
Thread.sleep(1000);
String s = "return string";
return s;
}
}
2. Executor框架成員
2.1 ThreadPoolExecutor
直接創建ThreadPoolExecutor的實體物件，見https://www.cnblogs.com/chiaki/p/13536624.html
ThreadPoolExecutor通常使用工廠類Executors創建，可以創建3種型別的ThreadPoolExecutor，即FixedThreadPool、SingleThreadExecutor以及CachedThreadPool，
-
FixedThreadPool：適用于為了滿足資源管理的需求，而需要限制當先執行緒數量的應用場景，適用于負載比較重的服務器，
public static ExecutorService es = Executors.newFixedThreadPool(int threadNums); public static ExecutorService es = Executors.newFixedThreadPool(int threadNums, ThreadFactory threadFactory);public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }把執行緒池最大執行緒數量maxmumPoolSize和核心執行緒池的數量corePoolSize設定為
threadNums，將引數keepAliveTime設定為0L，使用無界佇列LinkedBlockingQueue作為阻塞佇列，因此當任務不能立刻執行時，都會添加到阻塞佇列中，而且maximumPoolSize，keepAliveTime都是無效的， -
SingleThreadExecutor：適用于需要保證順序地執行各個任務；并且在任意時間點，不會有多個執行緒是活動地應用場景，**
public static ExecutorService es = Executors.newSingleThreadExecutor(); public static ExecutorService es = Executors.newSingleThreadExecutor(ThreadFactory threadFactory);public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); }因為阻塞佇列使用的是
LinkedBlockingQueue，因此和FixedThreadPool一樣，引數maximumPoolSize以及keepAliveTime都是無效的，corePoolSize為1，因此最多只能創建一個執行緒， -
CachedThreadPool：大小無界的執行緒池，適用于執行很多的短期異步任務的小程式，或者是負載較輕的服務器，
public static ExecutorService es = Executors.newCachedThreadPool(); public static ExecutorService es = Executors.newCachedThreadPool(ThreadFactory threadFactory);public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }CachedThreadPool使用
SynchronizedQueue作為阻塞佇列，SynchronizedQueue是不存盤元素的阻塞佇列，實作“一對一的交付”，也就是說，每次向佇列中put一個任務必須等有執行緒來take這個任務，否則就會一直阻塞該任務，如果一個執行緒要take一個任務就要一直阻塞知道有任務被put進阻塞佇列，因為CachedThreadPool的maximumPoolSize為
Integer.MUX_VALUE，因此CachedThreadPool是無界的執行緒池，也就是說可以一直不斷的創建執行緒，這樣可能會使CPU和記憶體資源耗盡，corePoolSize為0，因此在CachedThreadPool中直接通過阻塞佇列來進行任務的提交，
2.2 ScheduledThreadPoolExecutor
ScheduledThreadPoolExecutor類繼承了ThreadPoolExecutor并實作了ScheduledExecutorService介面，主要用于在給定的延遲后執行任務或者定期執行任務，
ScheduledThreadPoolExecutor通常使用Executors工廠類來創建，可創建2種型別的ScheduledThreadPoolExecutor，即ScheduledThreadPoolExecutor和SingleThreadScheduledExecutor，
-
ScheduledThreadPoolExecutor：適用于若干個（固定）執行緒延時或者定期執行任務，同時為了滿足資源管理的需求而需要限制后臺執行緒數量的場景，
public static ScheduledExecutorService ses = Executors.newScheduledThreadPool(int threadNums); public static ScheduledExecutorService ses = Executors.newScheduledThreadPool(int threadNums, ThreadFactory threadFactory); -
SingleThreadScheduledExecutor：適用于需要單個執行緒延時或者定期的執行任務，同時需要保證各個任務順序執行的應用場景，
public static ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor(int threadNums); public static ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor(int threadNums, ThreadFactory threadFactory);
ScheduledThreadPoolExecutor的實作：
ScheduledThreadPoolExecutor的實作主要是通過把任務封裝為ScheduledFutureTask來實作，通過呼叫scheduledAtFixedTime()方法或者scheduledWithFixedDelay()方法向阻塞佇列添加一個實作了RunnableScheduledFutureTask介面的ScheduledFutureTask類物件，ScheduledFutureTask主要包括3個成員變數：
// 序列號，用于保存任務添加到阻塞佇列的順序
private final long sequenceNumber;
// 用于保存該任務將要被執行的具體時間
private long time;
// 周期，用于保存任務直線的間隔周期
private final long period;
ScheduledTreadPoolExecutor的阻塞佇列是用無界佇列DelayQueue實作的，可以實作元素延時delayTime后才能獲取元素，在ScheduledThreadPoolExecutor中，DelayQueue內部封裝了一個PriorityQueue，來對任務進行排序，首先對time排序，time小的在前，如果time一樣，則sequence小的在前，也就是說如果time一樣，那么先被提交的任務先執行，
因為DelayQueue是一個無界的佇列，因此執行緒池的maximumPoolSize是無效的，ScheduledThreadPoolExecutor的作業流程大致如下：
2.3 Future介面/FutureTask實作類
Future介面和實作Future介面的FutureTask實作類，代表異步計算的結果，
2.3.1 FutureTask的使用
FutureTask除了實作Future介面外還實作了Runnable介面，因此，FutureTask可以交給Executor執行，也可以條用執行緒直接執行（FutureTask.run()），根據FutureTask.run()方法被執行的時機，FutureTask可處于以下3種狀態：
- 未啟動：創建了一個FutureTask物件但沒有執行FutureTask.run()；
- 已啟動：FutureTask.run()方法被執行的程序中；
- 已完成：FutureTask.run()正常執行結束，或者FutureTask被取消（
FutureTask.cancel()），或者執行FutureTask.run()時拋出例外而例外結束；
狀態遷移示意圖：
FutureTask的get和cancle執行示意圖：
2.3.2 FutureTask的實作
FutureTask是一個基于AQS同步佇列實作的一個自定義同步組件，通過對同步狀態state的競爭實作acquire或者release操作，
FutureTask的內部類Sync實作了AQS介面，通過對tryAcquire等抽象方法的重寫和模板方法的呼叫來實作內部類Sync的tryAcquireShared等方法，然后聚合Sync的方法來實作FutureTask的get和cancel等方法，
FutureTask的設計示意圖：
FutureTask的get方法最侄訓呼叫AQS.acquireSharedInterruptibly(int arg)方法：
- 呼叫
AQS.acquireSharedInterruptibly(int arg)方法會首先呼叫tryAcquireShared()方法判斷acquire操作是否可以成功，可以成功的條件是state為執行完成狀態RAN或者已取消狀態CANCELLED，且runner不為null； - 如果成功則
get()方法立即回傳，如果失敗則到執行緒等待佇列執行release操作； - 當其他執行緒執行release操作喚醒當前執行緒后（比如
FutureTask.run()或FutureTask.cancle(...)），當前執行緒再次執行tryAcquireShared()將回傳正值1，當前執行緒離開現場等待佇列并喚醒它的后繼執行緒（級聯喚醒）； - 最后返回計算的結果或拋出例外，
2.3.3 FutureTask的使用場景
- 當一個執行緒需要等待另一個執行緒把某個任務執行完以后它才能繼續執行時；
- 有若干執行緒執行若干任務，每個任務最多只能被執行一次；
- 當多個執行緒師徒執行同一個任務，但只能允許一個執行緒執行此任務，其它執行緒需要等這個任務被執行完畢以后才能繼續執行時，
2.4 Runnable和Callable介面
用于實作執行緒要執行的作業單元，
2.5 Executors工廠類
提供了常見配置執行緒池的方法，因為ThreadPoolExecutor的引數眾多且意義重大，為了避免配置出錯，才有了Executors工廠類，
3. 為什么不建議使用Executors創建執行緒池？
FixedThreadPool和SingleThreadExecutor：允許請求的佇列長度為Integer.MAX_VALUE（無界的阻塞佇列），可能堆積大量的請求，從而導致OOM，
CachedThreadPool和ScheduledThreadPool：允許創建的執行緒數量為Integer.MAX_VALUE（無界的阻塞佇列），可能會創建大量執行緒，從而導致OOM，
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/61809.html
標籤：Java
上一篇：MyEclipse2017的swing design不顯示圖片icon
下一篇：框架學習黑洞