掌握系列之并發編程-9.線程池

掌握高并發、高可用架構

第二課 并發編程

從本課開始學習并發編程的內容。主要介紹并發編程的基礎知識、鎖、內存模型、線程池、各種并發容器的使用。

第九節 線程池

線程池 Executors Executor ExecutorService ThreadPoolExecutor

為何要使用線程池

有以下幾個原因：

1. 線程池改進了多線程應用程序的響應時間：由于線程池中的線程已經準備好且等待被分配任務，可以直接拿來使用而不用新建一個線程
2. 線程池節省了為每個短生命周期任務而創建一個完整的線程開銷，并且可以在任務完成后回收資源
3. 線程池根據當前在系統中運行的進程來優化線程時間片
4. 線程池運行我們開啟多個任務而不用為每個線程單獨設置屬性
5. 線程池允許我們為正在執行的任務的程序參數傳遞一個包含狀態信息的對象引用
6. 線程池可以用來解決處理一個特定請求最大線程數量限制的問題

根本上說，我們使用線程池主要就是為了減少創建和銷毀線程的次數，每個線程 都可以重復利用，可以執行多個任務，從而節省內存（線程開的越多，消耗的內存越大），提高了資源利用率。

線程池ThreadPoolExecutor

代碼起始是Executor，這是一個接口。

抽象類AbstractExecutorService實現了接口ExecutorService，ExecutorService又繼承了Executor，AbstractExecutorService的默認實現類是ThreadPoolExecutor，這個類就是線程池。

ThreadPoolExecutor有4個構造函數：

public ThreadPoolExecutor(int corePoolSize,
                          int maximunPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Running> workQueue);

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Running> workQueue,
                          ThreadFactory factory);

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Running> workQueue,
                          RejectedExecutionHandler handler);

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Running> workQueue,
                          ThreadFactory factory,
                          RejectedExecutionHandler handler);

corePoolSize:線程池中核心線程的最大數量

核心線程，線程池創建后，如果當前線程總數小于corePoolSize，則新建的就是核心線程，如果超過corePoolSize，則新建的就是非核心線程。

核心線程默認情況下會一直存活在線程池中，即使這個核心線程處于閑置狀態。

如果指定ThreadPoolExecutor的allowCoreThreadTimeOut為true，那么核心線程如果處于閑置狀態的話，超過一定時間（keepAliveTime）也會被銷毀。

maximumPoolSize: 線程池中線程總數的最大值

線程總數 = 核心線程數 + 非核心線程數

keepAliveTime: 線程池中非核心線程閑置超時時長

默認情況下，一個非核心線程，如果閑置時長超過該參數設置，就會被銷毀。如果設置參數allowCoreThreadTimeOut為true，則超時時長也會作用于核心線程。

TimeUnit unit: 時長單位

枚舉值，MILLSECONDS：毫秒；SECONDS：秒；MINUTS：分鐘；HOURS：小時；DAYS：天

workQueue: 阻塞隊列

線程池中的任務隊列，維護著等待執行的Runnable對象。當所有的核心線程都在干活時，新添加的任務會被添加到這個隊列中等待處理，如果這個隊列滿了，則會新建非核心線程來執行任務。

public interface BlockingQueue<E> extends Queue<E> {
    // 將指定元素添加到隊列，成功返回true，否則拋出異常；如果是給限定了長度的隊列中添加元素，推薦offer
    boolean add(E e);
    // 將指定的元素添加的隊列，如果成功則返回true，否則返回false；元素不能為空，否則拋出NPE
    boolean offer(E e);
    // 將元素添加到隊列，如果隊列沒有多余空間，方法會一直阻塞，直到隊列有多余空間
    boolean put(E e) throws InterruptedException;
    // 將元素在指定的時間內添加到隊列中，成功返回true，否則返回false
    boolean off(E e, long timeout, TimeUnit unit) throws InterruptedException;
    // 從隊列中獲取值，如果隊列沒有元素，方法會一直阻塞，直到隊列有值
    E take() throws InterruptedException;
    // 在給定的時間內獲取隊列中的值，沒有獲取到會拋出異常
    E poll(long timeout, TimeUnit unit) throws InterruptedException;
    // 獲取隊列的剩余空間
    int remainingCapacity();
    // 移除指定的值
    boolean remove(Object o);
    // 判斷隊列是否包含值
    boolean contains(Object o);
    // 將隊列中的所有元素都移除，并設置到指定集合中
    int drainTo(Collection<? extends E> c);
}

一般來說，workQueue有以下四種隊列類型：

SynchronousQueue：同步隊列，這種隊列在接收到任務時，會直接提交給線程處理，而不會保留它。假如所有線程都在忙碌，則會新建線程來處理這個任務。所以為了防止出現線程數達到maximumPoolSize而不能新建線程的錯誤，當使用這種隊列時，需要把maximumPoolSize指定為Integer.MAX_VALUE。

LinkedBlockingQueue：×××鏈表阻塞隊列，這種隊列接收到任務時，如果當前線程數少于核心線程數，則會新建核心線程來處理任務；如果當前線程數達到核心線程數，則會保持到該隊列中；由于隊列×××，即所有超過核心線程數的任務都會存入隊列，所以會導致maximumPoolSize失效。

ArrayBlockingQueue：有界數組阻塞隊列，接收到任務時，如果當前線程數少于核心線程數，則會新建核心線程來處理任務；如果當前線程數達到核心線程數，則會保持到該隊列中；如果隊列已滿，則新建線程來執行任務，如果隊列已滿，而且線程數已達到maximumPoolSize指定的數量，則會報錯。

DelayQueue：延遲隊列，隊列元素必須實現Delayed接口，接收到任務時，先入隊列，只有達到了指定時間，才會執行任務。

ThreadFactory factory: 創建線程的方式

這是一個接口，通過調用它的方法： Thread newThread(Runnable r)來創建線程

RejectedExecutionHandler handler: 線程池無法創建線程時，如何拋出異常

一般是當線程池中的線程數量已經達到最大值，或者線程池已經關閉時，會拋出一個RejectedExecutionException

既然線程池新添加了任務，那么線程池是如何處理這些批量任務？

1. 如果線程數量未達到corePoolSize，則新建一個線程(核心線程)執行任務
2. 如果線程數量達到了corePoolSize，則將任務移入隊列等待
3. 如果隊列已滿，新建線程(非核心線程)執行任務
4. 如果隊列已滿，總線程數又達到了maximumPoolSize，就會由RejectedExecutionHandler拋出異常

四種線程池

1 newFixedThreadPool：定長的線程池，可控制線程的最大并發數，超出的任務會在隊列中等待。

public class Executors {
    public static ExecutorService newFixedThreadPool(int nThread) {
        return new ThreadPoolExecutor(nThread, nThread, 0L, TimeUnit.MILLSECONDS, new LinkedBlockingQueue<Runnable>());
    }
}

2 newCachedThreadPool：可緩存的線程池，如果線程池長度超過需要，可以靈活回收空閑線程，如無可回收線程，則會新建線程來處理任務。

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
}

3 newScheduledThreadPool：定長任務線程池，支持定時和周期性任務執行。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

4 newSingleThreadExecutor：一個 單線程的線程池，只有一個線程來執行任務。

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService(
        new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLSECONDS, new LinkedBlockingQueue<Runnable>());
    );
}