問題：

1.單機上一個執行緒正在處理服務，如果忽然斷電了怎麼辦（正在處理和阻塞佇列裡的請求怎麼處理）

2.為什麼要使用執行緒池，執行緒池用什麼用

3.說說幾種常見的執行緒池及使用場景

4.執行緒池有哪幾種工作佇列

5.怎麼理解無界佇列和有界佇列

6.執行緒池中的幾種重要的引數及流程

什麼是執行緒池

執行緒池的概念大家應該都很清楚，幫我們重複管理執行緒，避免建立大量的執行緒增加開銷。

除了降低開銷以外，執行緒池也可以提高響應速度，瞭解點 JVM 的同學可能知道，一個物件的建立大概需要經過以下幾步：

1. 檢查對應的類是否已經被載入、解析和初始化
2. 類載入後，為新生物件分配記憶體
3. 將分配到的記憶體空間初始為 0
4. 對物件進行關鍵資訊的設定，比如物件的雜湊碼等
5. 然後執行 init 方法初始化物件

建立一個物件的開銷需要經過這麼多步，也是需要時間的嘛，那可以複用已經建立好的執行緒的執行緒池，自然也在提高響應速度上做了貢獻。

java.util.concurrent.Executors提供了一個 java.util.concurrent.Executor介面的實現用於建立執行緒池

用執行緒池控制執行緒數量，其他執行緒排隊等候。一個任務執行完畢，再從佇列的中取最前面的任務開始執行。若佇列中沒有等待程序，執行緒池的這一資源處於等待。當一個新任務需要執行時，如果執行緒池中有等待的工作執行緒，就可以開始運行了；否則進入等待佇列。

執行緒池的應用場景

假設一個伺服器完成一項任務所需時間為：T1 建立執行緒時間，T2 線上程中執行任務的時間，T3 銷燬執行緒時間。如果：T1 + T3 遠大於 T2，則可以採用執行緒池，以提高伺服器效能。

執行緒池技術正是關注如何縮短或調整T1,T3時間的技術，從而提高伺服器程式效能的。它把T1，T3分別安排在伺服器程式的啟動和結束的時間段或者一些空閒的時間段，這樣在伺服器程式處理客戶請求時，不會有T1，T3的開銷了。

執行緒池不僅調整T1,T3產生的時間段，而且它還顯著減少了建立執行緒的數目，看一個例子：

假設一個伺服器一天要處理50000個請求，並且每個請求需要一個單獨的執行緒完成。線上程池中，執行緒數一般是固定的，所以產生執行緒總數不會超過執行緒池中執行緒的數目，而如果伺服器不利用執行緒池來處理這些請求則執行緒總數為50000。一般執行緒池大小是遠小於50000。所以利用執行緒池的伺服器程式不會為了建立50000而在處理請求時浪費時間，從而提高效率。

執行緒池的組成

一個執行緒池包括以下四個基本組成部分：

1、執行緒池管理器（ThreadPool）：用於建立並管理執行緒池，包括 建立執行緒池，銷燬執行緒池，新增新任務；

2、工作執行緒（PoolWorker）：執行緒池中執行緒，在沒有任務時處於等待狀態，可以迴圈的執行任務；

3、任務介面（Task）：每個任務必須實現的介面，以供工作執行緒排程任務的執行，它主要規定了任務的入口，任務執行完後的                                            收尾工作，任務的執行狀態等；

4、任務佇列（taskQueue）：用於存放沒有處理的任務。提供一種緩衝機制。

ThreadPoolExecutor類中其他的一些比較重要成員變數：

private final BlockingQueue<Runnable> workQueue;              //任務快取佇列，用來存放等待執行的任務
private final ReentrantLock mainLock = new ReentrantLock();   //執行緒池的主要狀態鎖，對執行緒池狀態（比如執行緒池大小
                                                              //、runState等）的改變都要使用這個鎖
private final HashSet<Worker> workers = new HashSet<Worker>();  //用來存放工作集
 
private volatile long  keepAliveTime;    //執行緒存活時間   
private volatile boolean allowCoreThreadTimeOut;   //是否允許為核心執行緒設定存活時間
private volatile int   corePoolSize;     //核心池的大小（即執行緒池中的執行緒數目大於這個引數時，提交的任務會被放進任務快取佇列）
private volatile int   maximumPoolSize;   //執行緒池最大能容忍的執行緒數
 
private volatile int   poolSize;       //執行緒池中當前的執行緒數
 
private volatile RejectedExecutionHandler handler; //任務拒絕策略
 
private volatile ThreadFactory threadFactory;   //執行緒工廠，用來建立執行緒
 
private int largestPoolSize;   //用來記錄執行緒池中曾經出現過的最大執行緒數
 
private long completedTaskCount;   //用來記錄已經執行完畢的任務個數

執行緒池的“真相”

Java裡面執行緒池的頂級介面是Executor，但是嚴格意義上講Executor並不是一個執行緒池，而只是一個執行執行緒的工具。真正的執行緒池介面是ExecutorService。

submit 方法：

submit 方法會提交一個任務去給執行緒池執行，該任務可以是帶返回結果的 Callable<V> 任務，也可以是一開始就指定結果的 Runnable 任務，或者不帶結果的 Runnable 任務（此時即一開始指定結果為 null）。submit 方法會返回一個與所提交任務相關聯的 Future<V>。通過 上一篇文章 我們可以知道，Future<V> 的 get 方法可以等待任務執行完畢並返回結果。所以通過 Future<V>，我們可以與已經提交到執行緒池的任務進行互動。submit 提交任務及任務執行過程大致如下：

1. 向執行緒池提交一個 Runnable 或者 Callable<V> 任務；
2. 將 任務 作為引數使用 newTaskFor 方法構造出 FutureTask<V>；通過 Future<V> 的 get 方法，獲得任務的結果。

invokeAll 方法：

invokeAll 方法可以一次執行多個任務，但它並不同等於多次呼叫 submit 方法。submit 方法是非阻塞的，每次呼叫 submit 方法提交任務到執行緒池之後，會立即返回與任務相關聯的 Future<V>，然後當前執行緒繼續向後執行；

而 invokeAll 方法是阻塞的，只有當提交的多個任務都執行完畢之後，invokeAll 方法才會返回，執行結果會以List<Future<V>>返回,該 List<Future<V>> 中的每個 Future<V> 是和提交任務時的 Collection<Callable<V>> 中的任務 Callable<V> 一 一對應的。帶 timeout 引數的 invokeAll 就是設定一個超時時間，如果超過這個時間 invokeAll 中提交的所有任務還有沒全部執行完，那麼沒有執行完的任務會被取消（中斷），之後同樣以一個 List<Future<V>> 返回執行的結果。

invokeAny 方法：

invokeAny 方法也是阻塞的，與 invokeAll 方法的不同之處在於，當所提交的一組任務中的任何一個任務完成之後，invokeAny 方法便會返回（返回的結果便是那個已經完成的任務的返回值），而其他任務會被取消（中斷）。

舉一個 invokeAny 使用的例子：電腦有 C、D、E、F 四個盤，我們需要找一個檔案，但是我們不知道這個檔案位於哪個盤中，我們便可以使用 invokeAny，並提交四個任務（對應於四個執行緒）分別查詢 C、D、E、F 四個盤，如果哪個執行緒找到了這個檔案，那麼此時 invokeAny 便停止阻塞並返回結果，同時取消其他任務。

shutdown 方法：

shutdown 方法的作用是向執行緒池傳送關閉的指令。一旦線上程池上呼叫 shutdown 方法之後，執行緒池便不能再接受新的任務；如果此時還向執行緒池提交任務，那麼將會丟擲 RejectedExecutionException 異常。之後執行緒池不會立刻關閉，直到之前已經提交到執行緒池中的所有任務（包括正在執行的任務和在佇列中等待的任務）都已經處理完成，才會關閉。

shutdownNow 方法：

與 shutdown 不同，shutdownNow 會立即關閉執行緒池 —— 當前線上程池中執行的任務會全部被取消，然後返回執行緒池中所有正在等待的任務。

（值得注意的是，我們 必須顯式的關閉執行緒池，否則執行緒池不會自己關閉）

awaitTermination 方法：

awaitTermination 可以用來判斷執行緒池是否已經關閉。呼叫 awaitTermination 之後，在 timeout 時間內，如果執行緒池沒有關閉，則阻塞當前執行緒，否則返回 true；當超過 timeout 的時間後，若執行緒池已經關閉則返回 true，否則返回 false。該方法一般這樣使用：

1. 任務全部提交完畢之後，我們呼叫 shutdown 方法向執行緒池傳送關閉的指令；
2. 然後我們通過 awaitTermination 來檢測到執行緒池是否已經關閉，可以得知執行緒池中所有的任務是否已經執行完畢；
3. 執行緒池執行完已經提交的所有任務，並將自己關閉；
4. 呼叫 awaitTermination 方法的執行緒停止阻塞，並返回 true；

isShutdown() 方法，如果執行緒池已經呼叫 shutdown 或者 shutdownNow，則返回 true，否則返回 false；

isTerminated() 方法，如果執行緒池已經呼叫 shutdown 並且執行緒池中所有的任務已經執行完畢，或者執行緒池呼叫了 shutdownNow，則返回 true，否則返回 false。

執行緒池狀態

在ThreadPoolExecutor中定義了一個volatile變數，另外定義了幾個static final變量表示執行緒池的各個狀態：

volatile int runState;
static final int RUNNING    = 0;
static final int SHUTDOWN   = 1;
static final int STOP       = 2;
static final int TERMINATED = 3;

 runState表示當前執行緒池的狀態，它是一個volatile變數用來保證執行緒之間的可見性；

　　下面的幾個static final變量表示runState可能的幾個取值。

　　當建立執行緒池後，初始時，執行緒池處於RUNNING狀態；

　　如果呼叫了shutdown()方法，則執行緒池處於SHUTDOWN狀態，此時執行緒池不能夠接受新的任務，它會等待所有任務執行完畢；

　　如果呼叫了shutdownNow()方法，則執行緒池處於STOP狀態，此時執行緒池不能接受新的任務，並且會去嘗試終止正在執行的任務；

　　當執行緒池處於SHUTDOWN或STOP狀態，並且所有工作執行緒已經銷燬，任務快取佇列已經清空或執行結束後，執行緒池被設定為TERMINATED狀態。

執行緒池的處理流程

建立執行緒池需要使用 ThreadPoolExecutor 類，它的建構函式引數如下：

public ThreadPoolExecutor(int corePoolSize,    //核心執行緒的數量
                          int maximumPoolSize,    //最大執行緒數量
                          long keepAliveTime,    //超出核心執行緒數量以外的執行緒空餘存活時間
                          TimeUnit unit,    //存活時間的單位
                          BlockingQueue<Runnable> workQueue,    //儲存待執行任務的佇列
                          ThreadFactory threadFactory,    //建立新執行緒使用的工廠
                          RejectedExecutionHandler handler // 當任務無法執行時的處理器
                          ) {...}

引數介紹如註釋所示，要了解這些引數左右著什麼，就需要了解執行緒池具體的執行方法ThreadPoolExecutor.execute:

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    int c = ctl.get();
    //1.當前池中執行緒比核心數少，新建一個執行緒執行任務
    if (workerCountOf(c) < corePoolSize) {   
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //2.核心池已滿，但任務佇列未滿，新增到佇列中
    if (isRunning(c) && workQueue.offer(command)) {   
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))    //如果這時被關閉了，拒絕任務
            reject(command);
        else if (workerCountOf(recheck) == 0)    //如果之前的執行緒已被銷燬完，新建一個執行緒
            addWorker(null, false);
    }
    //3.核心池已滿，佇列已滿，試著建立一個新執行緒
    else if (!addWorker(command, false))
        reject(command);    //如果建立新執行緒失敗了，說明執行緒池被關閉或者執行緒池完全滿了，拒絕任務
}

• 可以看到，執行緒池處理一個任務主要分三步處理，程式碼註釋裡已經介紹了，我再用通俗易懂的例子解釋一下：

（執行緒比作員工，執行緒池比作一個團隊，核心池比作團隊中核心團隊員工數，核心池外的比作外包員工）

1. 有了新需求，先看核心員工數量超沒超出最大核心員工數，還有名額的話就新招一個核心員工來做 
   
   • 需要獲取全域性鎖
2. 核心員工已經最多了，HR 不給批 HC 了，那這個需求只好攢著，放到待完成任務列表吧
3. 如果列表已經堆滿了，核心員工基本沒機會搞完這麼多工了，那就找個外包吧 
   
   • 需要獲取全域性鎖
4. 如果核心員工 + 外包員工的數量已經是團隊最多能承受人數了，沒辦法，這個需求接不了了

結合這張圖，這回流程你明白了嗎？

由於 1 和 3 新建執行緒時需要獲取全域性鎖，這將嚴重影響效能。因此 ThreadPoolExecutor 這樣的處理流程是為了在執行 execute() 方法時儘量少地執行 1 和 3，多執行 2。

在 ThreadPoolExecutor 完成預熱後（當前執行緒數不少於核心執行緒數），幾乎所有的 execute() 都是在執行步驟 2。

前面提到的 ThreadPoolExecutor 建構函式的引數，分別影響以下內容：

• corePoolSize：核心執行緒池數量 
  
  • 線上程數少於核心數量時，有新任務進來就新建一個執行緒，即使有的執行緒沒事幹
  • 等超出核心數量後，就不會新建執行緒了，空閒的執行緒就得去任務佇列裡取任務執行了
• maximumPoolSize：最大執行緒數量 
  
  • 包括核心執行緒池數量 + 核心以外的數量
  • 如果任務佇列滿了，並且池中執行緒數小於最大執行緒數且大於核心執行緒池數量，會再建立新的執行緒執行任務
• keepAliveTime：核心池以外的執行緒存活時間，即沒有任務的外包的存活時間 
  
  • 如果給執行緒池設定 allowCoreThreadTimeOut(true)，則核心執行緒在空閒時頭上也會響起死亡的倒計時
  • 如果任務是多而容易執行的，可以調大這個引數，那樣執行緒就可以在存活的時間裡有更大可能接受新任務
• workQueue：儲存待執行任務的阻塞佇列 
  
  • 不同的任務型別有不同的選擇，下一小節介紹
• threadFactory：每個執行緒建立的地方 
  
  • 可以給執行緒起個好聽的名字，設定個優先順序啥的
• handler：飽和策略，大家都很忙，咋辦呢，有四種策略 
  
  • CallerRunsPolicy：只要執行緒池沒關閉，就直接用呼叫者所線上程來執行任務
  • AbortPolicy：直接丟擲 RejectedExecutionException 異常
  • DiscardPolicy：悄悄把任務放生，不做了
  • DiscardOldestPolicy：把佇列裡待最久的那個任務扔了，然後再呼叫 execute() 試試看能行不
  • 我們也可以實現自己的 RejectedExecutionHandler 介面自定義策略，比如如記錄日誌什麼的

儲存待執行任務的阻塞佇列

當執行緒池中的核心執行緒數已滿時，任務就要儲存到佇列中了。

執行緒池中使用的佇列是 BlockingQueue 介面，常用的實現有如下幾種：

• ArrayBlockingQueue：  有界佇列，基於陣列、有界，按 FIFO（先進先出）原則對元素進行排序
• LinkedBlockingQueue：無界佇列，基於連結串列，按FIFO （先進先出） 排序元素 
  
  • 吞吐量通常要高於 ArrayBlockingQueue
  • Executors.newFixedThreadPool() 使用了這個佇列
• SynchronousQueue：直接提交，不儲存元素的阻塞佇列 
  
  • 每個插入操作必須等到另一個執行緒呼叫移除操作，否則插入操作一直處於阻塞狀態
  • 吞吐量通常要高於 LinkedBlockingQueue
  • Executors.newCachedThreadPool使用了這個佇列
• PriorityBlockingQueue：具有優先順序的、無限阻塞佇列

建立自己的執行緒池

瞭解上面的內容後，我們就可以建立自己的執行緒池了。

①先定義執行緒池的幾個關鍵屬性的值：

private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2; // 核心執行緒數為 CPU 數＊2
private static final int MAXIMUM_POOL_SIZE = 64;    // 執行緒池最大執行緒數
private static final int KEEP_ALIVE_TIME = 1;    // 保持存活時間 1秒

• 設定核心池的數量為 CPU 數的兩倍，一般是 4、8，好點的 16 個執行緒
• 最大執行緒數設定為 64
• 空閒執行緒的存活時間設定為 1 秒

②然後根據處理的任務型別選擇不同的阻塞佇列

如果是要求高吞吐量的，可以使用 SynchronousQueue 佇列；如果對執行順序有要求，可以使用 PriorityBlockingQueue；如果最大積攢的待做任務有上限，可以使用 LinkedBlockingQueue。

private final BlockingQueue<Runnable> mWorkQueue = new LinkedBlockingQueue<>(128);

③然後建立自己的 ThreadFactory

在其中為每個執行緒設定個名稱：

private final ThreadFactory DEFAULT_THREAD_FACTORY = new ThreadFactory() {
    private final AtomicInteger mCount = new AtomicInteger(1);

    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r, TAG + " #" + mCount.getAndIncrement());
        thread.setPriority(Thread.NORM_PRIORITY);
        return thread;
    }
};

④然後就可以建立執行緒池了

private ThreadPoolExecutor mExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME,
        TimeUnit.SECONDS, mWorkQueue, DEFAULT_THREAD_FACTORY,
        new ThreadPoolExecutor.DiscardOldestPolicy());

這裡我們選擇的飽和策略為 DiscardOldestPolicy，你可以可以建立自己的。

⑤完整程式碼：

public class ThreadPoolManager {
    private final String TAG = this.getClass().getSimpleName();
    private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2; // 核心執行緒數為 CPU數＊2
    private static final int MAXIMUM_POOL_SIZE = 64;    // 執行緒佇列最大執行緒數
    private static final int KEEP_ALIVE_TIME = 1;    // 保持存活時間 1秒

    private final BlockingQueue<Runnable> mWorkQueue = new LinkedBlockingQueue<>(128);

    private final ThreadFactory DEFAULT_THREAD_FACTORY = new ThreadFactory() {
        private final AtomicInteger mCount = new AtomicInteger(1);

        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r, TAG + " #" + mCount.getAndIncrement());
            thread.setPriority(Thread.NORM_PRIORITY);
            return thread;
        }
    };

    private ThreadPoolExecutor mExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME,
            TimeUnit.SECONDS, mWorkQueue, DEFAULT_THREAD_FACTORY,
            new ThreadPoolExecutor.DiscardOldestPolicy());

    private static volatile ThreadPoolManager mInstance = new ThreadPoolManager();

    public static ThreadPoolManager getInstance() {
        return mInstance;
    }

    public void addTask(Runnable runnable) {
        mExecutor.execute(runnable);
    }

    @Deprecated
    public void shutdownNow() {
        mExecutor.shutdownNow();
    }
}

這樣我們就有了自己的執行緒池。

JDK 提供的執行緒池及使用場景

JDK 為我們內建了五種常見執行緒池的實現，均可以使用 Executors 工廠類建立。

1.newFixedThreadPool（固定大小的執行緒池）

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

不招外包，有固定數量核心成員的正常網際網路團隊。

可以看到，FixedThreadPool 的核心執行緒數和最大執行緒數都是指定值，也就是說當執行緒池中的執行緒數超過核心執行緒數後，任務都會被放到阻塞佇列中。

此外 keepAliveTime 為 0，也就是多餘的空餘執行緒會被立即終止（由於這裡沒有多餘執行緒，這個引數也沒什麼意義了）。

而這裡選用的阻塞佇列是 LinkedBlockingQueue，使用的是預設容量 Integer.MAX_VALUE，相當於沒有上限。

因此這個執行緒池執行任務的流程如下：

1. 執行緒數少於核心執行緒數，也就是設定的執行緒數時，新建執行緒執行任務
2. 執行緒數等於核心執行緒數後，將任務加入阻塞佇列 
   
   • 由於佇列容量非常大，可以一直加加加
3. 執行完任務的執行緒反覆去佇列中取任務執行

FixedThreadPool 用於負載比較重的伺服器，為了資源的合理利用，需要限制當前執行緒數量。

相關推薦