【併發編程】【JDK源碼】J.U.C--線程池

原文：慕課網實戰·高併發探索（十四）：線程池 Executor

new Thread的弊端

每次new Thread 新建對象，性能差。
線程缺少統一管理，可能無限制的新建線程，相互競爭，可能佔用過多的系統資源致使死機或者OOM（out of memory 內存溢出），這種問題的緣由不是由於單純的new一個Thread，而是可能由於程序的bug或者設計上的缺陷致使不斷new Thread形成的。
缺乏更多功能，如更多執行、按期執行、線程中斷。

線程池的好處

重用存在的線程，減小對象建立、消亡的開銷，性能好。
可有效控制最大併發線程數，提升系統資源利用率，同時能夠避免過多資源競爭，避免阻塞。
提供定時執行、按期執行、單線程、併發數控制等功能。

線程池類圖

在線程池的類圖中，咱們最常使用的是最下邊的Executors,用它來建立線程池使用線程。那麼在上邊的類圖中，包含了一個Executor框架，它是一個根據一組執行策略的調用調度執行和控制異步任務的框架，目的是提供一種將任務提交與任務如何運行分離開的機制。它包含了三個executor接口：

Executor:運行新任務的簡單接口
ExecutorService：擴展了Executor，添加了用來管理執行器生命週期和任務生命週期的方法
ScheduleExcutorService：擴展了ExecutorService，支持Future和按期執行任務。

線程池核心類-ThreadPoolExecutor

參數說明：

ThreadPoolExecutor一共有七個參數，這七個參數配合起來，構成了線程池強大的功能。

corePoolSize：核心線程數量
maximumPoolSize：線程最大線程數
workQueue：阻塞隊列，存儲等待執行的任務，很重要，會對線程池運行過程產生重大影響

當咱們提交一個新的任務到線程池，線程池會根據當前池中正在運行的線程數量來決定該任務的處理方式。處理方式有三種：
一、直接切換（SynchronusQueue）
二、無界隊列（LinkedBlockingQueue）可以建立的最大線程數爲corePoolSize,這時maximumPoolSize就不會起做用了。當線程池中全部的核心線程都是運行狀態的時候，新的任務提交就會放入等待隊列中。
三、有界隊列（ArrayBlockingQueue）最大maximumPoolSize，可以下降資源消耗，可是這種方式使得線程池對線程調度變的更困難。由於線程池與隊列容量都是有限的。因此想讓線程池的吞吐率和處理任務達到一個合理的範圍，又想使咱們的線程調度相對簡單，而且還儘量下降資源的消耗，咱們就須要合理的限制這兩個數量
分配技巧： [若是想下降資源的消耗包括下降cpu使用率、操做系統資源的消耗、上下文切換的開銷等等，能夠設置一個較大的隊列容量和較小的線程池容量，這樣會下降線程池的吞吐量。若是咱們提交的任務常常發生阻塞，咱們能夠調整maximumPoolSize。若是咱們的隊列容量較小，咱們須要把線程池大小設置的大一些，這樣cpu的使用率相對來講會高一些。可是若是線程池的容量設置的過大，提升任務的數量過多的時候，併發量會增長，那麼線程之間的調度就是一個須要考慮的問題。這樣反而可能會下降處理任務的吞吐量。

keepAliveTime：線程沒有任務執行時最多保持多久時間終止（當線程中的線程數量大於corePoolSize的時候，若是這時沒有新的任務提交核心線程外的線程不會當即銷燬，而是等待，直到超過keepAliveTime）
unit：keepAliveTime的時間單位
threadFactory：線程工廠，用來建立線程，有一個默認的工場來建立線程，這樣新建立出來的線程有相同的優先級，是非守護線程、設置好了名稱
rejectHandler：當拒絕處理任務時(阻塞隊列滿)的策略（AbortPolicy默認策略直接拋出異常、CallerRunsPolicy用調用者所在的線程執行任務、DiscardOldestPolicy丟棄隊列中最靠前的任務並執行當前任務、DiscardPolicy直接丟棄當前任務）

corePoolSize、maximumPoolSize、workQueue 三者關係：若是運行的線程數小於corePoolSize的時候，直接建立新線程來處理任務。即便線程池中的其餘線程是空閒的。若是運行中的線程數大於corePoolSize且小於maximumPoolSize時，那麼只有當workQueue滿的時候才建立新的線程去處理任務。若是corePoolSize與maximumPoolSize是相同的，那麼建立的線程池大小是固定的。這時有新任務提交，當workQueue未滿時，就把請求放入workQueue中。等待空線程從workQueue取出任務。若是workQueue此時也滿了，那麼就使用另外的拒絕策略參數去執行拒絕策略。

初始化方法：由七個參數組合成四個初始化方法

其餘方法：

序號	方法名	描述
1	execute()	提交任務，交給線程池執行
2	submit()	提交任務，可以返回執行結果 execute+Future
3	shutdown()	關閉線程池，等待任務都執行完
4	shutdownNow()	關閉線程池，不等待任務執行完
5	getTaskCount()	線程池已執行和未執行的任務總數
6	getCompleteTaskCount()	已完成的任務數量
7	getPoolSize()	線程池當前的線程數量
8	getActiveCount()	當前線程池中正在執行任務的線程數量

線程池生命週期：

running：能接受新提交的任務，也能處理阻塞隊列中的任務
shutdown：不能處理新的任務，可是能繼續處理阻塞隊列中任務
stop：不能接收新的任務，也不處理隊列中的任務
tidying：若是全部的任務都已經終止了，這時有效線程數爲0
terminated：最終狀態

使用Executor建立線程池

使用Executor能夠建立四種線程池：分別對應上邊提到的四種線程池初始化方法。

Executors.newCachedThreadPool
Executors.newFixedThreadPool
Executors.newSingleThreadExecutor
Executors.newScheduledThreadPool

一、Executors.newCachedThreadPool

建立一個可緩存的線程池，若是線程池的長度超過了處理的須要，能夠靈活回收空閒線程。若是沒有可回收的就新建線程。

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

//使用方法：
public static void main(String[] args) {
    ExecutorService executorService = Executors.newCachedThreadPool();
    for (int i = 0; i < 10; i++) {
        final int index = i;
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                log.info("task:{}", index);
            }
        });
    }
    executorService.shutdown();
}

值得注意的一點是，newCachedThreadPool的返回值是ExecutorService類型，該類型只包含基礎的線程池方法，但卻不包含線程監控相關方法，所以在使用返回值爲ExecutorService的線程池類型建立新線程時要考慮到具體狀況。

二、Executors.newFixedThreadPool

定長線程池，能夠線程現成的最大併發數，超出在隊列等待

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

//使用方法：
public static void main(String[] args) {
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    for (int i = 0; i < 10; i++) {
        final int index = i;
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                log.info("task:{}", index);
            }
        });
    }
    executorService.shutdown();
}

三、Executors.newSingleThreadExecutor

單線程化的線程池，用惟一的一個共用線程執行任務，保證全部任務按指定順序執行（FIFO、優先級…）。

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

//使用方法：
public static void main(String[] args) {
    ExecutorService executorService = Executors.newSingleThreadExecutor();
    for (int i = 0; i < 10; i++) {
        final int index = i;
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                log.info("task:{}", index);
            }
        });
    }
    executorService.shutdown();
}

四、Executors.newScheduledThreadPool

定長線程池，支持定時和週期任務執行。

//源碼：
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,//此處super指的是ThreadPoolExecutor
          new DelayedWorkQueue());
}

//基礎使用方法：
public static void main(String[] args) {
    ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
    executorService.schedule(new Runnable() {
        @Override
        public void run() {
            log.warn("schedule run");
        }
    }, 3, TimeUnit.SECONDS);//延遲3秒執行
    executorService.shutdown();
}

ScheduledExecutorService提供了三種方法可使用：

scheduleAtFixedRate：以指定的速率執行任務
scheduleWithFixedDelay：以指定的延遲執行任務

舉例：

executorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        log.warn("schedule run");
    }
}, 1, 3, TimeUnit.SECONDS);//延遲一秒後每隔3秒執行

小擴展：延遲執行任務的操做，java中還有Timer類一樣能夠實現。

Timer timer = new Timer();
timer.schedule(new TimerTask() {
    @Override
    public void run() {
        log.warn("timer run");
    }
}, new Date(), 5 * 1000);

參考資料：
深刻理解java線程池—ThreadPoolExecutor