Java多執行緒之Executor框架和手寫簡易的執行緒池
目錄
執行緒池
什麼是執行緒池
執行緒池一種執行緒使用模式，執行緒池會維護多個執行緒，等待著分配可併發執行的任務，當有任務需要執行緒執行時，從執行緒池中分配執行緒給該任務而不用主動的建立執行緒。
執行緒池的好處
如果在我們平時如果需要用到執行緒時，我們一般是這樣做的：建立執行緒(T1)，使用建立的執行緒來執行任務(T2)，任務執行完成後銷燬當前執行緒(T3)，這三個階段是必須要有的。
而如果使用執行緒池呢？
執行緒池會預先建立好一定數量的執行緒，需要的時候申請使用，在一個任務執行完後也不需要將該執行緒銷燬，很明顯的節省了T1和T3這兩階段的時間。
同時我們的執行緒由執行緒池來統一進行管理，這樣也提高了執行緒的可管理性。
手寫一個自己的執行緒池
現在我們可以簡單的理解為執行緒池實際上就是存放多個執行緒的陣列，在程式啟動是預先例項化一定得執行緒例項，當有任務需要時分配出去。現在我們先來寫一個自己的執行緒池來理解一下執行緒池基本的工作過程。
執行緒池需要些什麼？
首先執行緒池肯定需要一定數量的執行緒，所以首先需要一個執行緒陣列，當然也可以是一個集合。
執行緒陣列是用來進行存放執行緒例項的，要使用這些執行緒就需要有任務提交過來。當任務量過大時，我們是不可能在同一時刻給所有的任務分配一個執行緒的，所以我們還需要一個用於存放任務的容器
這裡的預先初始化執行緒例項的數量也需要我們來根據業務確定。
同時執行緒例項的數量也不能隨意的定義，所以我們還需要設定一個最大執行緒數。
//執行緒池中允許的最大執行緒數 private static int MAXTHREDNUM = Integer.MAX_VALUE; //當用戶沒有指定時預設的執行緒數 private int threadNum = 6; //執行緒佇列，存放執行緒任務 private List<Runnable> queue; private WorkerThread[] workerThreads;
執行緒池工作
執行緒池的執行緒一般需要預先進行例項化，這裡我們通過建構函式來模擬這個過程。
public MyThreadPool(int threadNum) {
this.threadNum = threadNum;
if(threadNum > MAXTHREDNUM)
threadNum = MAXTHREDNUM;
this.queue = new LinkedList<>();
this.workerThreads = new WorkerThread[threadNum];
init();
}
//初始化執行緒池中的執行緒
private void init(){
for(int i=0;i<threadNum;i++){
workerThreads[i] = new WorkerThread();
workerThreads[i].start();
}
}線上程池準備好了後，我們需要像執行緒池中提交工作任務，任務統一提交到佇列中，當有任務時，自動分發執行緒。
//提交任務
public void execute(Runnable task){
synchronized (queue){
queue.add(task);
//提交任務後喚醒等待在佇列的執行緒
queue.notifyAll();
}
}我們的工作執行緒為了獲取任務，需要一直監聽任務佇列，當佇列中有任務時就由一個執行緒去執行，這裡我們用到了前面提到的安全中斷。
private class WorkerThread extends Thread {
private volatile boolean on = true;
@Override
public void run() {
Runnable task = null;
//判斷是否可以取任務
try {
while(on&&!isInterrupted()){
synchronized (queue){
while (on && !isInterrupted() && queue.isEmpty()) {
//這裡如果使用阻塞佇列來獲取在執行時就不會報錯
//報錯是因為退出時銷燬了所有的執行緒資源，不影響使用
queue.wait(1000);
}
if (on && !isInterrupted() && !queue.isEmpty()) {
task = queue.remove(0);
}
if(task !=null){
//取到任務後執行
task.run();
}
}
}
} catch (InterruptedException e) {
e.printStackTrace();
}
task = null;//任務結束後手動置空，加速回收
}
public void cancel(){
on = false;
interrupt();
}
}當然退出時還需要對執行緒池中的執行緒等進行銷燬。
//銷燬執行緒池
public void shutdown(){
for(int i=0;i<threadNum;i++){
workerThreads[i].cancel();
workerThreads[i] = null;
}
queue.clear();
}好了，到這裡我們的一個簡易版的執行緒池就完成了，功能雖然不多但是執行緒池執行的基本原理差不多實現了，實際上非常簡單，我們來寫個程式測試一下：
public class ThreadPoolTest {
public static void main(String[] args) throws InterruptedException {
// 建立3個執行緒的執行緒池
MyThreadPool t = new MyThreadPool(3);
CountDownLatch countDownLatch = new CountDownLatch(5);
t.execute(new MyTask(countDownLatch, "testA"));
t.execute(new MyTask(countDownLatch, "testB"));
t.execute(new MyTask(countDownLatch, "testC"));
t.execute(new MyTask(countDownLatch, "testD"));
t.execute(new MyTask(countDownLatch, "testE"));
countDownLatch.await();
Thread.sleep(500);
t.shutdown();// 所有執行緒都執行完成才destory
System.out.println("finished...");
}
// 任務類
static class MyTask implements Runnable {
private CountDownLatch countDownLatch;
private String name;
private Random r = new Random();
public MyTask(CountDownLatch countDownLatch, String name) {
this.countDownLatch = countDownLatch;
this.name = name;
}
public String getName() {
return name;
}
@Override
public void run() {// 執行任務
try {
countDownLatch.countDown();
Thread.sleep(r.nextInt(1000));
System.out.println("任務 " + name + " 完成");
} catch (InterruptedException e) {
System.out.println(Thread.currentThread().getId()+" sleep InterruptedException:"
+Thread.currentThread().isInterrupted());
}
}
}
}
result：
任務 testA 完成
任務 testB 完成
任務 testC 完成
任務 testD 完成
任務 testE 完成
finished...
java.lang.InterruptedException
at java.lang.Object.wait(Native Method)
at com.learn.threadpool.MyThreadPool$WorkerThread.run(MyThreadPool.java:75)
...從結果可以看到我們提交的任務都被執行了，當所有任務執行完成後，我們強制銷燬了所有執行緒，所以會丟擲異常。
JDK中的執行緒池
上面我們實現了一個簡易的執行緒池，稍微理解執行緒池的基本運作原理。現在我們來認識一些JDK中提供了執行緒池吧。
ThreadPoolExecutor
public class ThreadPoolExecutor extends AbstractExecutorServiceThreadPoolExecutor是一個ExecutorService ，使用可能的幾個合併的執行緒執行每個提交的任務，通常使用Executors工廠方法配置，通過Executors可以配置多種適合不同場景的執行緒池。
ThreadPoolExecutor中的主要引數
ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) corePoolSize
執行緒池中的核心執行緒數，當外部提交一個任務時，執行緒池就建立一個新執行緒執行任務，直到當前執行緒數等於corePoolSize時不再建立新執行緒；
如果當前執行緒數為corePoolSize，繼續提交的任務被儲存到阻塞佇列中，等待被執行；
如果執行了執行緒池的prestartAllCoreThreads()方法，執行緒池會提前建立並啟動所有核心執行緒。
maximumPoolSize
執行緒池中允許的最大執行緒數。如果當前阻塞佇列已滿，還在繼續提交任務，則建立新的執行緒執行任務，前提是當前執行緒數小於maximumPoolSize。
keepAliveTime
執行緒空閒時的存活時間，即當執行緒沒有任務執行時，繼續存活的時間。預設情況下，執行緒一般不會被銷燬，該引數只在執行緒數大於corePoolSize時才有用。
workQueue
workQueue必須是阻塞佇列。當執行緒池中的執行緒數超過corePoolSize的時候，執行緒會進入阻塞佇列進行等待。阻塞佇列可以使有界的也可以是無界的。
threadFactory
建立執行緒的工廠，通過自定義的執行緒工廠可以給每個新建的執行緒設定一個執行緒名。Executors靜態工廠裡預設的threadFactory，執行緒的命名規則是“pool-{數字}-thread-{數字}”。
RejectedExecutionHandler
執行緒池的飽和處理策略，當阻塞佇列滿了，且沒有空閒的工作執行緒，如果繼續提交任務，必須採取一種策略處理該任務，執行緒池提供了4種策略：
- AbortPolicy：直接丟擲異常，預設的處理策略
- CallerRunsPolicy：使用呼叫者所屬的執行緒來執行當前任務
- DiscardOldestPolicy：丟棄阻塞佇列中靠最前的任務，並執行當前任務
- DiscardPolicy：直接丟棄該任務
如果上述提供的處理策略無法滿足業務需求，也可以根據場景實現RejectedExecutionHandler介面，自定義飽和策略，如記錄日誌或持久化儲存不能處理的任務。
ThreadPoolExecutor中的主要執行流程
//圖片來自網路
- 執行緒池判斷核心執行緒池裡的執行緒(corePoolSize)是否都在執行任務。如果不是，則建立一個新的工作執行緒來執行任務。如果核心執行緒池裡的執行緒都在執行任務，則進入2。
- 執行緒池判斷工作佇列(workQueue)是否已滿。如果工作佇列沒有滿，則將新提交的任務儲存在該佇列裡。如果工作佇列滿了，則進入3。
- 執行緒池判斷執行緒池的執行緒(maximumPoolSize)是否都處於工作狀態。如果沒有，則建立一個新的工作執行緒來執行任務。如果已經滿了，則交給飽和策略來處理這個任務。
這裡需要注意的是核心執行緒池大小指得是corePoolSize引數，而執行緒池工作執行緒數指的是maximumPoolSize。
Executor
實際上我們在使用執行緒池時，並不一定需要自己來定義上面介紹的引數的值，JDK為我們提供了一個排程框架。通過這個排程框架我們可以輕鬆的建立好執行緒池以及非同步的獲取任務的執行結果。
排程框架的組成
任務
一般是指需要被執行的任務，多為使用者提供。被提交的任務需要實現Runnable介面或Callable介面。
任務的執行
Executor是任務執行機制的核心介面，其將任務的提交和執行分離開來。ExecutorService繼承了Executor並做了一些擴充套件，可以產生Future為跟蹤一個或多個非同步任務執行。任務的執行主要是通過實現了Executor和ExecutorService介面的類來進行實現。例如：ThreadPoolExecutor和ScheduledThreadPoolExecutor。
結果獲取
對結果的獲取可以通過Future介面以及其子類介面來實現。Future介面提供了一系列諸如檢查是否就緒，是否執行完成，阻塞以及獲取結果等方法。
Executors工廠中的執行緒池
FixedThreadPool
new ThreadPoolExecutor(nThreads, nThreads, 0L,
TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());該執行緒池中corePoolSize和maximumPoolSize引數一致。同時使用無界阻塞佇列，將會導致maximumPoolSize和keepAliveTime已經飽和策略無效，因為佇列會一直接收任務，直到OOM。
SingleThreadExecutor
new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>())該執行緒池中corePoolSize和maximumPoolSize都為1，表示始終只有一個執行緒在工作，適用於需要保證順序地執行各個任務；並且在任意時間點，不會有多個執行緒是活動的應用場景。同時使用無界阻塞佇列，當任務多時極有可能OOM。
CachedThreadPool
new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>()CachedThreadPool型別的執行緒池corePoolSize為0，表示任務將會提交給佇列，但是SynchronousQueue又是一個不包含任何容量的佇列。所以每一個任務提交過來都會建立一個新的執行緒來執行，該型別的執行緒池適用於執行很多的短期非同步任務的程式，或者是負載較輕的伺服器。如果當任務的提交速度一旦超過任務的執行速度，在極端情況下可能會因為建立過多執行緒而耗盡CPU和記憶體資源。
ScheduledThreadPool
對於定時任務型別的執行緒池，Executor可以建立兩種不同型別的執行緒池：ScheduledThreadPoolExecutor和SingleThreadScheduledExecutor，前者是包含若干個執行緒的ScheduledThreadPoolExecutor，後者是隻包含一個的ScheduledThreadPoolExecutor。
ScheduledThreadPoolExecutor適用於需要多個後臺執行緒執行週期任務，同時為了滿足資源管理的需求而需要限制後臺執行緒的數量的應用場景。
SingleThreadScheduledExecutor適用於需要單個後臺執行緒執行週期任務，同時需要保證順序地執行各個任務的應用場景。
new ThreadPoolExecutor(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue());在對該型別執行緒池進行例項化時，我們可以看到maximumPoolSize設定為了Integer的最大值，所以很明顯在極端情況下和CachedThreadPool型別一樣可能會因為建立過多執行緒而耗盡CPU和記憶體資源。
DelayedWorkQueue是一種延時阻塞佇列，此佇列的特點為其中元素只能在其延遲到期時才被使用。ScheduledThreadPool型別在執行任務時和其他執行緒池有些不同。
- ScheduledThreadPool型別執行緒池中的執行緒(假設現線上程A開始取任務)從DelayedWorkQueue中取已經到期的任務。
- 執行緒A獲取到任務後開始執行。
- 任務執行完成後設定該任務下一次執行的時間。
- 將該任務重新放入到執行緒池中。
ScheduledThreadPool中存在著定時任務和延時任務兩種。
延時任務通過schedule(...)方法以及過載方法和scheduleWithFixedDelay實現，延時任務通過設定某個時間間隔後執行，schedule(...)僅執行一次。
定時任務由scheduleAtFixedRate實現。該方法建立並執行在給定的初始延遲之後，隨後以給定的時間段進行週期性動作，即固定時間間隔的任務。
特殊的scheduleWithFixedDelay方法是建立並執行在給定的初始延遲之後首先啟用的定期動作，隨後在一個執行的終止和下一個執行的開始之間給定的延遲，即固定延時間隔的任務。
固定時間間隔的任務不論每次任務花費多少時間，下次任務開始執行時間是確定的。對於scheduleAtFixedRate方法中，若任務處理時長超出設定的定時頻率時長,本次任務執行完才開始下次任務,下次任務已經處於超時狀態,會馬上開始執行。若任務處理時長小於定時頻率時長,任務執行完後,定時器等待,下次任務會在定時器等待頻率時長後執行。
固定延時間隔的任務是指每次執行完任務以後都等待一個固定的時間。由於作業系統排程以及每次任務執行的語句可能不同，所以每次任務執行所花費的時間是不確定的，也就導致了每次任務的執行週期存在一定的波動。
需要注意的是定時或延時任務中所涉及到時間、週期不能保證實時性及準確性，實際執行中會有一定的誤差。
Callable/Future
在介紹實現多執行緒的時候我們有簡單介紹過Runnable和Callable的，這兩者基本相同，不同在於Callable可以返回一個結果，而Runnable不返回結果。對於Callable介面的使用方法和Runnable基本相同，同時我們也可以選擇是否對結果進行接收處理。在Executors中提供了將Runnable轉換為Callable的api：Callable<Object> callable(Runnable task)。
Future是一個用於接收Runnable和Callable計算結果的介面，當然它還提供了查詢任務狀態，中斷或者阻塞任務以及查詢結果的能力。
boolean cancel(boolean mayInterruptIfRunning) //嘗試取消執行此任務。
V get() //等待計算完成，然後檢索其結果。
V get(long timeout, TimeUnit unit) //等待最多在給定的時間，然後檢索其結果（如果可用）。
boolean isCancelled() //如果此任務在正常完成之前被取消，則返回 true 。
boolean isDone() //如果任務已完成返回true。 FutureTask是對Future的基本實現，具有啟動和取消計算的方法，查詢計算是否完整，並檢索計算結果。FutureTask對Future做了一定得擴充套件：
void run() //將此future設定為其計算結果，除非已被取消。
protected boolean runAndReset() //執行計算而不設定其結果，然後重置為初始狀態，如果計算遇到異常或被取消，則不執行此操作。
protected void set(V v) //將此Future的結果設定為給定值，除非Future已被設定或已被取消。
protected void setException(Throwable t) //除非已經設定了此 Future 或已將其取消，否則它將報告一個 ExecutionException，並將給定的 throwable 作為其原因。 FutureTask除了實現Future介面外，還實現了Runnable介面。所以FutureTask可以由Executor執行，也可以由呼叫執行緒直接執行futureTask.run()。
當FutureTask處於未啟動或已啟動狀態時，執行FutureTask.get()方法將導致呼叫執行緒阻塞；
當FutureTask處於已完成狀態時，執行FutureTask.get()方法將導致呼叫執行緒立即返回結果或丟擲異常。
當FutureTask處於未啟動狀態時，執行FutureTask.cancel()方法將導致此任務永遠不會被執行；
當FutureTask處於已啟動狀態時，執行FutureTask.cancel(true)方法將以中斷執行此任務執行緒的方式來嘗試停止該任務；
當FutureTask處於已啟動狀態時，執行FutureTask.cancel(false)方法將不會對正在執行此任務的執行緒產生影響（讓正在執行的任務執行完成）。
關於是否使用Executors
在之前阿里巴巴出的java開發手冊中，有明確提出禁止使用Executors：
【強制】執行緒池不允許使用 Executors 去建立，而是通過 ThreadPoolExecutor 的方式，
這樣的處理方式讓寫的同學更加明確執行緒池的執行規則，規避資源耗盡的風險。
在上面我們分析過使用Executors建立的幾種執行緒池的使用場景和缺點，大多數情況下出問題在於可能導致OOM，在我實際使用中基本沒有遇到過這樣的情況。但是考慮到阿里巴巴這樣體量的併發請求，可能遇到這種情況的機率較大。所以我們還是應該根據實際情況考慮是否使用，當然實際遵循阿里巴巴開發手冊來可能會更好一點，畢竟這是國類頂尖公司常年在生產中積累下的經驗。
最後，在本節中只是簡單介紹執行緒池及其基本原理，幫助更好的理解執行緒池。並不涉及具體如何使用。