多執行緒的軟體設計，確實可以最大限度的發揮現代多核處理器的計算能力，提高系統的吞吐量和效能。但是，若不加控制和管理的隨意使用執行緒，對系統的效能反而會產生不利的影響。

一、什麼是執行緒池

為了避免系統頻繁的建立和銷燬執行緒，我們可以讓建立的執行緒複用。如果你寫過資料庫方便的程式碼，那你肯定知道資料庫連線池。或者是傳送http請求時候的，http連線池。亦或者是傳送mq訊息時候的連線池。執行緒池於他們的作用是一樣的，執行緒池中，總是有幾個活躍的執行緒。當需要執行緒的時候，從執行緒池中拿一個空閒的來用，用完了再放進去。

二、JDK對執行緒池的支援

為了能夠更好的控制多執行緒，JDK提供了一套Executor框架，幫助開發人員進行有效的執行緒控制，其本質就是一個執行緒池。
Executor框架提供了各種型別的執行緒池，主要有一下幾種：

• newFixedThreadPool()：該方法返回一個固定執行緒數量的執行緒池。該執行緒池中的執行緒數量始終保持不變。當有一個新的任務提交時，執行緒池中若有空閒執行緒，則立即執行，若沒有，則新的任務被暫存在一個佇列中，待有執行緒空閒時，便處理在任務佇列中的任務。
• newSingleThreadExecutor()：該方法返回只有一個執行緒的執行緒池，若有多於一個的任務，被提交到執行緒池時，任務將會被暫存在一個佇列中，待執行緒池空閒時，按先入先出的順序執行佇列中的任務。
• newCachedThreadPool()：該方法返回一個可根據實際情況調整執行緒數量的執行緒池。執行緒池的數量不確定，但若有空閒執行緒可以複用，則會優先使用可複用執行緒。若所有執行緒均在工作，又有新任務提交，則會建立新的執行緒處理任務。所有執行緒，在當前任務執行完畢後，將返回執行緒池進行復用。
• newSingleThreadScheduledExecutor() ：該方法返回一個ScheduledExecutorService物件，執行緒池大小為1。ScheduledExecutorService介面在ExecutorService介面之上進行了擴充套件，在給定時間，執行某任務的功能，例如在固定的，某個延時之後執行，或者週期性執行某個任務。
• newScheduledThreadPool()：該方法也返回一個ScheduledExecutorService物件，但該執行緒池，可以指定執行緒數量。

1. 固定大小的執行緒池
這裡以newFixedThreadPool()為例，簡單展示下用法

package com.example.thread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by mazhenhua on 2017/3/9.
 */
public class ThreadPoolDemo {

    public static class MyTask implements Runnable{
        @Override
        public void run() {
            System.out.println(System.currentTimeMillis() + ":Thread ID:" + Thread.currentThread().getId());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e){
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        MyTask task = new MyTask();
        ExecutorService es = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 10; i++){
            es.submit(task);
        }


    }
}

輸出結果：

從這個結果中，我們看到，一共是5個執行緒，共執行了10次，當打出前5個執行緒的時候，停頓了1秒，然後又打印出了5個，並且全部列印完，程式並沒有結束，說明執行緒池還有活躍的執行緒。

2. 計劃任務

另外一個值得注意的方法是newScheduledThreadPool()，它返回一個ScheduledExecutorService物件，可以根據時間需要對執行緒進行排程。

ScheduledExecutorService與其他幾個執行緒池不同，它並不會立即安排執行任務。它其實是起到了定時任務的作用。它會在指定的時間，對任務進行排程。

public ScheduledFuture<?> schedule(Runnable command,
                                       long delay, TimeUnit unit)
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);

schedule()方法會在指定的時間，對任務進行一次排程。scheduleAtFixedRate()和scheduleWithFixedDelay()方法，會對任務進行週期性的呼叫，兩者的不同點是，FixedRate是從上一個任務開始後的period時間啟動下一次任務，而FixedDelay則是從上一個執行緒執行結束後delay時間後啟動下一個執行緒。一個是從上一個執行緒的開始時間計時，一個是從上一個執行緒的結束時間計時。

package com.example.thread;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * Created by mazhenhua on 2017/3/9.
 */
public class ScheduledExecutorServiceDemo {

    public static void main(String[] args) {
        ScheduledExecutorService ses = Executors.newScheduledThreadPool(10);
        ses.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(1000);
                    System.out.println(System.currentTimeMillis()/1000);
                } catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
        }, 10, 2, TimeUnit.SECONDS);
    }
}

執行上面程式碼可以發現，先停頓了10秒，然後每2秒列印一次。

上面的程式碼中，執行時sleep了1秒，所以週期為2秒一次的打印出來了，那麼如果sleep5秒會怎麼樣呢？
這裡不再囉嗦，直接給出結論，程式碼可以自己手動改，結論：會以5秒為週期的列印，任務並不會重疊而是在上次執行完畢後立即執行下一個任務。

如果是，scheduleWithFixedDelay方法就不一樣了，因為他本身就是上一個任務執行完畢才開始計時的，所以，如果是scheduleWithFixedDelay方法的話，那就會是7秒一個週期的列印，

排程程式實際上並不能保證任務會無限期的被持續呼叫。如果任務本身丟擲了異常，那麼後續的所有執行都會被中斷，因此，如果你想讓你任務持續穩定的執行，那麼一定要處理好異常。

3. 執行緒池的內部實現

對於核心的幾個執行緒池，無論是newFixedThreadPool()，newSingleThreadExecutor()還是newCachedThreadPool()，雖然看起來建立執行緒有著不同的特點，但是其內部實現均使用了，ThreadPoolExecutor：

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

• corePoolSize：指定了執行緒池中的執行緒數量。
• maximumPoolSize：指定了執行緒池中的最大執行緒數量。
• keepAliveTime：當執行緒池數量超過corePoolSize時，對於的空閒執行緒的存活時間。即，超過corePoolSize的執行緒，在多長時間內會被銷燬。
• unit：keepAliveTime的時間單位。
• workQueue：任務佇列，被提交但尚未被執行的佇列。
• threadFactory：執行緒工廠，用於建立執行緒，一般用預設的即可。
• handler：拒絕策略。當任務太多來不及處理時，如何拒絕任務。

這裡詳細說明一下workQueue和handler兩個引數，

引數workQueue指被提交，但未執行的任務佇列，他是一個BlockingQueue介面物件，僅存放Runable物件。根據佇列功能分類，在ThreadPoolExecutor的建構函式中，可以使用以下幾種BlockingQueue。

• 直接提交佇列：該功能由SynchronousQueue物件提供。SynchronousQueue是一個特殊的BlockingQueue。SynchronousQueue沒有容量，每一個插入操作，都要等待相應的刪除操作，反之，每一個刪除操作都要等待相應的插入操作。如果使用SynchronousQueue，提交的任務不會被真正的儲存，而總是將新任務提交給執行緒執行，如果沒有空執行緒，則嘗試建立執行緒，如果執行緒數量已經達到最大值，則執行拒絕策略。因此使用SynchronousQueue佇列，通常要設定maximumPoolSize值，否則很容執行拒絕策略。

• 有界的任務佇列：有界的任務佇列可以使用ArrayBlockingQueue實現，ArrayBlockingQueue的建構函式必須帶一個容量引數，表示該佇列的最大容量，當使用有界的任務佇列時，若有新的任務需要執行，如果執行緒池的實際執行緒數小於corePoolSize，則會優先建立新執行緒，若大於corePoolSize，則會將新任務加入等待佇列，若等待佇列已滿，無法加入，則在匯流排程數不大於maximumPoolSize的前提下，建立新的執行緒執行任務，若大於maximumPoolSize，則執行拒絕策略。可見，有界佇列僅當任務佇列裝滿時，才能將執行緒數提升到corePoolSize之上，換句話說，除非系統非常繁忙，否則確保核心執行緒數維持在corePoolSize

• 無界的任務佇列：無界任務佇列可以通過LinkedBlockingQueue類來實現。與有界佇列相比，除非系統資源耗盡，否則無界的任務佇列不存在任務入隊失敗的情況。當有新的任務到來，系統的執行緒數小於corePoolSize時，執行緒池會生成新的執行緒執行任務，但當系統的執行緒數達到corePoolSize後，就不會繼續增加。若後續又有任務繼續提交，而有沒有可用的執行緒時，則任務直接進入等待。若任務的建立和處理速度相差很大，無界佇列會保持快速增長，知道耗盡系統記憶體。

• 優先任務佇列：優先任務佇列是帶有執行優先順序的佇列。他通過PriorityBlockingQueue實現，可以控制任務的執行先後順序。他是一個特殊的無界佇列。無論是有界佇列ArrayBlockingQueue還是LinkedBlockingQueue無界佇列，都是按照先進先出的演算法處理任務的。而PriorityBlockingQueue則可以根據自身的優先順序順序先後執行，在確保系統性能的同時，也能有很好的質量保證。

newFixedThreadPool()方法返回了一個corePoolSize和maximumPoolSize一樣大小的，並使用了LinkedBlockingQueue無邊界的任務佇列執行緒池。因為對於固定大小的執行緒池，不存線上程數量大小的變化。

newSingleThreadExecutor()與newFixedThreadPool()類似只是corePoolSize和maximumPoolSize設定為了1

newCachedThreadPool()方法返回一個corePoolSize為0，maximumPoolSize無限大的執行緒池，這意味著，在沒有任務時，執行緒池內是沒有執行緒的，當有執行緒提交時，該執行緒池會使用空閒的執行緒執行任務，若無空閒執行緒，則將任務加入SynchronousQueue佇列，SynchronousQueue佇列是直接提交的佇列，它總會破使執行緒池增加新的執行緒執行任務，當執行完畢後，由於corePoolSize為0，因此空執行緒會在指定的時間（60s）內被回收。

下面是ThreadPoolExecutor的一段執行的核心程式碼

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

        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();

        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);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

程式碼中workerCountOf()方法獲取了當前執行緒池的執行緒總數。當執行緒總數小於corePoolSize核心執行緒數時，會將任務通過addWorker()方法直接排程執行。否則在workQueue.offer(command)方法進入等待佇列。如果等待佇列失敗(比如有界佇列達到上限，或者使用SynchronousQueue佇列)則會直接將任務提交給執行緒池，如果提交失敗，則執行拒絕策略。

4. 拒絕策略

上面多次提到了拒絕策略，那麼拒絕策略到底是什麼呢，請往下看。

拒絕策略就是當任務數量超過系統實際承載能力時，該如何處理。
JDK內建提供了四種策略：

• AbortPolicy策略：該策略會直接丟擲異常，阻止系統正常工作
• CallerRunsPolicy策略：只要執行緒池未關閉，該策略直接在呼叫執行緒中，運行當前被丟棄的任務。顯然這樣做不會真的丟棄任務，但是，任務提交執行緒的效能極有可能會急劇下降。
• DiscardPolicy策略：該策略，默默的丟棄無法處理的任務，不予任何處理。如果允許任務丟失，可以採用這個。
• DiscardOldestPolicy策略：該策略將丟棄最老的一個請求，也就是即將被執行的一個任務，並嘗試，再次提交這個任務。

以上策略都實現了RejectedExecutionHandler介面，若上述策略無法滿足需要，可以自己擴充套件RejectedExecutionHandler介面，

package com.example.thread;

import java.util.concurrent.*;

/**
 * Created by mazhenhua on 2017/3/13.
 */
public class RejectedThreadPoolDemo {

    public static class MyTask implements Runnable{

        @Override
        public void run() {
            System.out.println(System.currentTimeMillis() + ": Thread ID:" +
                Thread.currentThread().getId());
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        MyTask task = new MyTask();
        ExecutorService es = new ThreadPoolExecutor(5,5,0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingDeque<Runnable>(10),
                Executors.defaultThreadFactory(),
                new RejectedExecutionHandler(){
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        System.out.println(r.toString() + "is disCard");
                    }
                });
        for (int i = 0; i < Integer.MAX_VALUE; i ++){
            es.submit(task);
            Thread.sleep(10);
        }
    }
}

輸出結果：

上述程式碼中，執行緒池中有5個常駐執行緒，並且最大執行緒數也是5個，但是他卻擁有隻有10個容量的等待佇列。
MyTask的執行時間是100毫秒。所以當任務被源源不斷的提交時，拒絕策略就開始生效了。

5. 自定義執行緒建立：ThreadFactory

看了那麼多關於執行緒池的介紹，有沒有考慮一個基本的問題，執行緒池中的執行緒是哪裡來的？

執行緒池的作用是為了執行緒的複用，避免頻繁的建立執行緒，但是最開始的執行緒是哪裡開的呢？答案是：ThreadFactory

ThreadFactory介面只有一個方法建立執行緒Thread newThread(Runnable r)，當需要建立執行緒時，就呼叫這個方法。自定義執行緒，可以幫助我們做不少事。我們可以追蹤執行緒池在何時建立了多少個執行緒，也可以定義執行緒的組，優先順序，名字等等。也可以任性的將所有執行緒都設定為守護執行緒。

package com.example.thread;

import java.util.concurrent.*;

/**
 * Created by mazhenhua on 2017/3/13.
 */
public class RejectedThreadPoolDemo {

    public static class MyTask implements Runnable{

        @Override
        public void run() {
            //System.out.println(System.currentTimeMillis() + ": Thread ID:" +
            //    Thread.currentThread().getId());
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    //  拒絕策略
/*    public static void main(String[] args) throws InterruptedException {
        MyTask task = new MyTask();
        ExecutorService es = new ThreadPoolExecutor(5,5,0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingDeque<Runnable>(10),
                Executors.defaultThreadFactory(),
                new RejectedExecutionHandler(){
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        System.out.println(r.toString() + "is disCard");
                    }
                });
        for (int i = 0; i < Integer.MAX_VALUE; i ++){
            es.submit(task);
            Thread.sleep(10);
        }
    }*/

    public static void main(String[] args) throws InterruptedException{
        MyTask task = new MyTask();
        ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
                new SynchronousQueue<Runnable>(),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread t = new Thread(r);

                        t.setDaemon(true);

                        System.out.println("create : " + t);

                        return t;
                    }
                });


        for (int i = 0; i < 5; i ++){
            es.submit(task);
        }
        Thread.sleep(2000);
    }


}

執行結果：

從輸出結果中可以看出，建立了5個守護執行緒，並且當主執行緒main結束後，所有的執行緒都結束了。

6. 擴充套件執行緒池

雖然JDK已經幫我們實現了這個穩定的高效能執行緒池。但有的時候我們需要對這個執行緒池做一些擴充套件，比如，想要監控每個任務的執行開始時間結束時間，或者列印一些關鍵引數等等。

ThreadPoolExecutor提供了beforeExecute()，afterExecute()，terminated()三個介面對執行緒池進行控制，有點類似於spring的AOP

package com.example.thread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by mazhenhua on 2017/3/14.
 */
public class ExtThreadPool {

    public static class MyTask implements Runnable {
        public  String name;

        public MyTask(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            System.out.println("正在執行：Thread ID：" + Thread.currentThread().getId()
                + ", Task Name = " + name);
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService es = new ThreadPoolExecutor(5,5,0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()){
            @Override
            protected void beforeExecute(Thread t, Runnable r) {
                System.out.println("正在準備執行：" + ((MyTask) r).name);
            }

            @Override
            protected void afterExecute(Runnable r, Throwable t) {
                System.out.println("執行完成：" + ((MyTask) r).name);
            }

            @Override
            protected void terminated() {
                System.out.println("執行緒池退出");
            }
        };

        for (int i = 0; i < 5; i ++){
            MyTask task = new MyTask("TASK-" + i);
            es.execute(task);
            Thread.sleep(10);
        }
        es.shutdown();
    }

}

輸出結果：