java併發專題之線程池深刻淺出

1.線程池是什麼

線程池（Thread Pool）是一種基於池化思想管理線程的工具，常常出如今多線程服務器中，如MySQL。

咱們都知道線程的建立和銷燬都須要必定的資源開銷，下降了計算機的總體性能。那麼有沒有一種辦法能避免頻繁的線程建立和銷燬呢？基於此就引出了線程池的概念，使用線程池能夠帶來一系列好處：

• 下降資源消耗：經過池化技術重複利用已建立的線程，下降線程建立和銷燬形成的損耗。
• 提升響應速度：任務到達時，無需等待線程建立便可當即執行。
• 提升線程的可管理性：線程是稀缺資源，若是無限制建立，不只會消耗系統資源，還會由於線程的不合理分佈致使資源調度失衡，下降系統的穩定性。使用線程池能夠進行統一的分配、調優和監控。
• 提供更多更強大的功能：線程池具有可拓展性，容許開發人員向其中增長更多的功能。好比延時定時線程池ScheduledThreadPoolExecutor，就容許任務延期執行或按期執行。

2.線程池的設計

Java中JDK8的線程池核心實現類是ThreadPoolExecutor，咱們首先來看一下ThreadPoolExecutor的UML類圖，瞭解下ThreadPoolExecutor的繼承關係。

ThreadPoolExecutor實現的頂層接口是Executor，頂層接口Executor提供了一種思想：將任務提交和任務執行進行解耦。用戶無需關注如何建立線程，如何調度線程來執行任務，用戶只需提供Runnable對象，將任務的運行邏輯提交到執行器(Executor)中，由Executor框架完成線程的調配和任務的執行部分。ExecutorService接口增長了一些能力：（1）擴充執行任務的能力，補充能夠爲一個或一批異步任務生成Future的方法；（2）提供了管控線程池的方法，好比中止線程池的運行。AbstractExecutorService則是上層的抽象類，將執行任務的流程串聯了起來，保證下層的實現只需關注一個執行任務的方法便可。最下層的實現類ThreadPoolExecutor實現最複雜的運行部分，ThreadPoolExecutor將會一方面維護自身的生命週期，另外一方面同時管理線程和任務，使二者良好的結合從而執行並行任務。

3.線程池的實現

3.1. 經過Executors提供四種線程池：

一、newCachedThreadPool建立一個可緩存線程池，若是線程池長度超過處理須要，可靈活回收線程，若無可回收，則新建線程；線程池無限大，當執行第二個任務時第一個任務已完成，會複用執行第一個任務的線程，而不是新建線程。

二、newFixedThreadPool建立一個定長線程池，可控制線程最大併發數，超出的線程會在隊列中等待；初始線程數和最大線程數同樣，若是要執行的線程大於初始線程數，則會將多餘的線程任務加入到緩存隊列中等待執行。

三、newScheduledThreadPool建立一個定長線程池，支持定時及週期性任務的執行；

四、newSingleThreadExecutor建立一個單線程化的線程池，它只會用惟一的工做線程來執行任務，保證全部任務按照指定順序(FIFO,LIFO,優先級)執行；

經過Executors建立的線程池有個致命缺點，以newCachedThreadPool來講

public class ThreadPoolDemo {
    public static void main(String[] args)throws Exception {
        ExecutorService threadPool = Executors.newCachedThreadPool();
    }
}
複製代碼

咱們點進newCachedThreadPool()方法會看到以下內容：

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}
//咱們發現第二個參數爲Integer.MAX_VALUE，也就是最大的工做線程數爲Integer.MAX_VALUE，若是超高併發過來會直接OOM。
複製代碼

因此咱們得經過new ThreadPoolExecutor()來本身指定參數。

3.2經過ThreadPoolExecutor來建立線程池

構造方法以下：

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) 複製代碼

• corePoolSize：表示核心線程的數量，就是線程池中最好要保證的線程數量。
• maximumPoolSize：最大線程數量，就是當任務不少時，能工做的最大線程數。
• keepAliveTime：線程的空閒時間。就是當實際工做的線程數小於最大線程數的時候，就有部分線程是處於空閒的狀態，當這些空閒線程的空閒時間到達keepAliveTime就會被幹掉。可是要保證最少線程數爲corePoolSize。
• unit：空閒時間都單位。
• workQueue：工做隊列，就是當工做線程數達到corePoolSize以後的任務會放入到工做隊列中。
• threadFactory：線程工廠，用於建立線程。
• handler：拒絕處理器，就是當工做線程到達最大工做線程而且工做隊列已經滿了狀況下，線程池應該怎麼作。

3.3線程池的工做流程。

流程圖以下：

當任務過來時，首先會先去判斷線程池中工做的線程數量是否到達核心線程數，若是沒達到就直接執行，若是達到了就查看工做隊列是否滿。若是沒滿就將任務放入到工做隊列中，若是滿了就增長工做線程數來處理任務。若是工做線程和隊列都滿了的話就會用制定的策略去拒絕任務。

3.4 拒絕策略

• AbortPolicy(默認)：直接拋出異常RejectedExecutionException異常阻止系統正常運行。

• CallerRunsPolicy：調用者運行是一種調節機制，該策略既不會拋棄任務，也不會拋棄異常，而是將某些任務回退到調用者，從而下降新任務的流量。

• DiscardOldestPolicy：丟棄隊列中等待最久的任務，而後把當前任務加入隊列中嘗試再次提交當前任務。

• DiscardPolicy：直接丟棄任務，不予處理也不拋出異常。若是容許任務丟失，這是最好的一種方案。

代碼演示拒絕策略：

1.AbortPolicy

public static void main(String[] args)throws Exception {
    ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2,5,1L,TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(3),Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());
    try {
        for (int i = 0; i < 10; i++) {
            poolExecutor.execute(()->{
                System.out.println(Thread.currentThread().getName()+"\t辦理業務");
            });
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        poolExecutor.shutdown();
    }
    System.out.println(Thread.currentThread().getName()+"\t辦理業務");
}
複製代碼

也就是說明當工做線程和工做隊列都滿了以後線程池會拒絕任務直接報錯。

2.CallerRunsPolicy

public static void main(String[] args)throws Exception {
    ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2,5,1L,TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(3),Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.CallerRunsPolicy());
    try {
        for (int i = 0; i < 10; i++) {
            poolExecutor.execute(()->{
                System.out.println(Thread.currentThread().getName()+"\t辦理業務");
            });
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        poolExecutor.shutdown();
    }
    System.out.println(Thread.currentThread().getName()+"\t辦理業務");
}
複製代碼

也就是說明當工做線程和工做隊列都滿了以後會將任務返還給調用線程池的人，讓他去處理。

3.DiscardOldestPolicy和DiscardPolicy

public static void main(String[] args)throws Exception {
    ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2,5,1L,TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(3),Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.DiscardOldestPolicy());
    try {
        for (int i = 0; i < 10; i++) {
            poolExecutor.execute(()->{
                System.out.println(Thread.currentThread().getName()+"\t辦理業務");
            });
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        poolExecutor.shutdown();
    }
    System.out.println(Thread.currentThread().getName()+"\t辦理業務");
}
複製代碼

一共輸出了10條記錄，說明有一條消息被拋棄了。