講真 這次絕對讓你輕鬆學習線程池

本文轉載自【微信公眾號：五角錢的程式設計師，ID:xianglin965】經微信公眾號授權轉載，如需轉載與原文作者聯繫

5分鐘了解線程池

老王 是個深耕在帝都的一線碼農，辛苦一年掙了點錢，想把錢存儲到銀行卡裡，錢銀行卡辦理遇到了如下的遭遇

老王 銀行門口取號後發現有櫃檯營業但是沒人辦理業務直接辦理了。老王 取號後發現櫃檯都有人在辦理，等待席有空地，去坐著等辦理去了。老王 取號後發現櫃檯都有人辦理，等待席也人坐滿了，這個時候銀行經理看到小麥是老實人本著關愛老實人的態度，新開一個臨時窗口給他辦理了。老王 取號後發現櫃檯都滿了，等待座位席也滿了，臨時窗口也人滿了。這個時候銀行經理給出了若干解決策略。直接告知人太多不給你辦理了。看到老王 就來氣，也不給不辦理也不讓他走。經理讓老王 取嘗試跟座位席中最前面的人聊一聊看是否可以加塞，可以就辦理，不可以還是被踢走。經理直接跟老王 說誰讓你來的你找誰去我這辦理不了。

上面的這個流程幾乎就跟JDK線程池的大致流程類似，

營業中的3個窗口對應核心線程池數：corePoolSize銀行總的營業窗口數對應：maximumPoolSize打開的臨時窗口在多少時間內無人辦理則關閉對應：unit銀行裡的等待座椅就是等待隊列：workQueue無法辦理的時候銀行給出的解決方法對應：RejectedExecutionHandlerthreadFactory 該參數在JDK中是 線程工廠，用來創建線程對象，一般不會動。

5分鐘線程池的核心工作流程講解完畢，更細節的知識看下面。

什麼是線程池

簡單理解就是 預先創建好一定數量的線程對象，存入緩衝池中，需要用的時候直接從緩衝池中取出，用完之後不要銷毀，還回到緩衝池中。

線程池存在必要性

提高線程的利用率，降低資源的消耗。提高響應速度，線程的創建時間為T1，執行時間T2，銷毀時間T3，用線程池可以免去T1和T3的時間。便於統一管理線程對象可控制最大並發數

手動實現

如果先不看線程池源碼讓我們自己手動實現一個線程池你可以考慮到幾個重要點？

有若干個初始化好的線程數組來充當線程池。線程池要去一個 等待的任務隊列 中去拿任務。

簡單來說就是初始化N個線程充當線程池然後一起去阻塞隊列中進行阻塞take,新添加的任務都通過put將任務追加到任務隊列，關於任務隊列的講解看這blog

核心類

public class MyThreadPool2 {// 線程池中默認線程的個數為5 private static int WORK_NUM = 5; // 隊列默認任務個數為100 來不及保存任務 private static int TASK_COUNT = 100; // 工作線程組 private WorkThread[] workThreads; // 任務隊列，作為一個緩衝 private final BlockingQueue<Runnable> taskQueue; //用戶在構造這個池，希望的啟動的線程數 private final int worker_num; // 創建具有默認線程個數的線程池 public MyThreadPool2() { this(WORK_NUM, TASK_COUNT); } // 創建線程池,worker_num為線程池中工作線程的個數 public MyThreadPool2(int worker_num, int taskCount) { if (worker_num <= 0) worker_num = WORK_NUM; if (taskCount <= 0) taskCount = TASK_COUNT; this.worker_num = worker_num; taskQueue = new ArrayBlockingQueue<>(taskCount); workThreads = new WorkThread[worker_num]; for (int i = 0; i < worker_num; i++) { workThreads[i] = new WorkThread(); workThreads[i].start(); } Runtime.getRuntime().availableProcessors(); } // 執行任務,其實只是把任務加入任務隊列，什麼時候執行有線程池管理器決定 public void execute(Runnable task) { try { taskQueue.put(task);// 阻塞 放置任務 } catch (InterruptedException e) { e.printStackTrace(); } } // 銷毀線程池,該方法保證在所有任務都完成的情況下才銷毀所有線程，否則等待任務完成才銷毀 public void destroy() { // 工作線程停止工作，且置為null System.out.println("準備關閉線程池"); for (int i = 0; i < worker_num; i++) { workThreads[i].stopWorker(); workThreads[i] = null;//help gc } taskQueue.clear();// 清空任務隊列 } // 覆蓋toString方法，返回線程池信息：工作線程個數和已完成任務個數 @Override public String toString() { return "線程池大小 :" + worker_num + " 等待執行任務個數:" + taskQueue.size(); } //內部類，工作線程 private class WorkThread extends Thread { @Override public void run() { Runnable r = null; try { while (!isInterrupted()) { r = taskQueue.take();//阻塞獲得任務 if (r != null) { System.out.println(getId() + " 準備執行 :" + r); r.run(); } r = null; //help gc; } } catch (Exception e) { // TODO: handle exception } } public void stopWorker() { interrupt(); } }}

測試類

public class TestMyThreadPool {public static void main(String[] args) throws InterruptedException { // 創建3個線程的線程池 MyThreadPool2 t = new MyThreadPool2(3, 5); t.execute(new MyTask("testA")); t.execute(new MyTask("testB")); t.execute(new MyTask("testC")); t.execute(new MyTask("testD")); t.execute(new MyTask("testE")); System.out.println(t); Thread.sleep(10000); t.destroy();// 所有線程都執行完成 才destory System.out.println(t); } // 任務類 static class MyTask implements Runnable { private String name; private Random r = new Random(); public MyTask(String name) { this.name = name; } public String getName() { return name; } @Override public void run() {// 執行任務 try { Thread.sleep(r.nextInt(1000) + 2000); //隨機休眠 } catch (InterruptedException e) { System.out.println(Thread.currentThread().getId() + " 被打斷:" + Thread.currentThread().isInterrupted()); } System.out.println("任務 " + name + " 完成"); } }}

ThreadPoolExecutor

是JDK中所有線程池實現類的父類，構造函數有多個入參通過靈活的組合來實現線程池的初始化，核心構造函數如下。

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

重要參數解析

corePoolSize此值是用來初始化線程池中核心線程數，當線程池中線程池數<時，系統默認是添加一個任務才創建一個線程池。可以通過調用prestartAllCoreThreads方法一次性的啟動個數的線程。當線程數 = corePoolSize時，新任務會追加到workQueue中。

maximumPoolSize表示允許的最大線程數 = (非核心線程數+核心線程數)，當BlockingQueue也滿了，但線程池中總線程數 <時候就會再次創建新的線程。

keepAliveTime非核心線程 =(maximumPoolSize - corePoolSize ) ,非核心線程閒置下來不幹活最多存活時間。

unit線程池中非核心線程保持存活的時間TimeUnit.DAYS; 天 TimeUnit.HOURS; 小時 TimeUnit.MINUTES; 分鐘 TimeUnit.SECONDS; 秒 TimeUnit.MILLISECONDS; 毫秒 TimeUnit.MICROSECONDS; 微秒 TimeUnit.NANOSECONDS; 納秒

workQueue線程池 等待隊列，維護著等待執行的Runnable對象。當運行當線程數= corePoolSize時，新的任務會被添加到中，如果也滿了則嘗試用非核心線程執行任務，另外等待隊列儘量用有界的哦！！

threadFactory創建一個新線程時使用的工廠，可以用來設定線程名、是否為daemon線程等等。

handler、、都不可用的時候執行的 飽和策略。AbortPolicy ：直接拋出異常，默認用此 CallerRunsPolicy：用調用者所在的線程來執行任務 DiscardOldestPolicy：丟棄阻塞隊列裡最老的任務，隊列裡最靠前的任務 DiscardPolicy ：當前任務直接丟棄 想實現自己的飽和策略，實現RejectedExecutionHandler接口即可

形象流程圖如下：

提交

execute 不需要返回

// 核心思想跟上面的流程圖類似public void execute(Runnable command) { if (command == null) //規範性檢查 throw new NullPointerException(); int c = ctl.get();//當前工作的線程數跟線程狀態 ctl = AtomicInteger CAS級別 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) // 如果當前線程池中工作線程數小於核心線程數，直接添加任務 然後return return; c = ctl.get();// 添加失敗了重新獲得線程池中工作線程數 } if (isRunning(c) && workQueue.offer(command)) { // 線程池狀態是否處於可用，可用就嘗試將線程添加到queue int recheck = ctl.get();// 獲得線程池狀態 if (! isRunning(recheck) && remove(command)) reject(command);// 如果線程狀態不在運行中 則remove 該任務 else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false))// 嘗試將任務用非核心線程執行， reject(command);//失敗了則執行失敗策略。 }

submit 需要返回值ThreadPoolExecutor extends AbstractExecutorService父類中存在一個submit方法， public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); execute(ftask); return ftask;}關閉線程池

注意線程之間是協作式的哦，所以的關閉只是發出關閉指令。

shutdown() 將線程池狀態置為shutdown,並不會立即停止：停止接收外部submit的任務內部正在跑的任務和隊列裡等待的任務，會執行完等到第二步完成後，才真正停止

shutdownNow() 將線程池狀態置為stop。企圖立即停止，事實上不一定：跟shutdown()一樣，先停止接收外部提交的任務忽略隊列裡等待的任務嘗試將正在跑的任務interrupt中斷返回未執行的任務列表

shutdown 跟shutdownnow簡單來說區別如下：

shutdownNow()能立即停止線程池，正在跑的和正在等待的任務都停下了。這樣做立即生效，但是風險也比較大。shutdown()只是關閉了提交通道，用submit()是無效的；而內部該怎麼跑還是怎麼跑，跑完再停。

awaitTermination

pool.showdown()boolean b = pool.awaitTermination(3, TimeUnit.SECONDS)

有兩個參數，一個是timeout即超時時間，另一個是unit即時間單位。這個方法會使線程等待timeout時長，當超過timeout時間後，會監測

ExecutorService

是否已經關閉，若關閉則返回true，否則返回false。一般情況下會和shutdown方法組合使用，調用後當前線程會阻塞，直到

等所有已提交的任務（包括正在跑的和隊列中等待的）執行完或者等超時時間到或者線程被中斷，拋出InterruptedException

總結

優雅的關閉，用shutdown() 想立馬關閉，並得到未執行任務列表，用shutdownNow() 優雅的關閉，發出關閉指令後看下是否真的關閉了用awaitTermination()。

合理配置線程池

線程在Java中屬於稀缺資源，線程池不是越大越好也不是越小越好。任務分為計算密集型、IO密集型、混合型。

計算密集型：大部分都在用CPU跟內存，加密，邏輯操作業務處理等。IO密集型：資料庫連結，網絡通訊傳輸等。

計算密集型一般推薦線程池不要過大，一般是CPU數 + 1，+1是因為可能存在頁缺失(就是可能存在有些數據在硬碟中需要多來一個線程將數據讀入內存)。如果線程池數太大，可能會頻繁的 進行線程上下文切換跟任務調度。獲得當前CPU核心數代碼如下：

Runtime.getRuntime().availableProcessors();

IO密集型：線程數適當大一點，機器的Cpu核心數*2。混合型：如果密集型站大頭則拆分的必要性不大，如果IO型佔據不少有必要，Mark 下。常見線程池

每個線程池都是一個實現了接口

並且繼承自

的具體實現類，這些類的創建統一由一個工廠類

Executors

來提供對外創建接口。Executors框架圖如下：

中一個線程就是一個

Worker

對象，它與一個線程綁定，當

執行完畢就是線程執行完畢。而Worker帶了鎖

AQS

，根據我後面準備寫的讀寫鎖的例子，發現線程池是線程安全的。看看圖二的類圖。下面簡單介紹幾個常用的線程池模式。

FixedThreadPool

定長的線程池，有核心線程，核心線程的即為最大的線程數量，沒有非核心線程。使用的無界的等待隊列是LinkedBlockingQueue。使用時候小心堵滿等待隊列。

SingleThreadPool

只有一條線程來執行任務，適用於有順序的任務的應用場景，也是用的

無

界等待隊列

CachedThreadPool

可緩存的線程池，該線程池中沒有核心線程，非核心線程的數量為Integer.max_value，就是無限大，當有需要時創建線程來執行任務，沒有需要時回收線程，適用於耗時少，任務量大的情況。任務隊列用的是SynchronousQueue如果生產多快消費慢，則會導致創建很多線程需注意。

WorkStealingPool

JDK7以後基於ForkJoinPool實現。

PS：其中

、

都用的

等待隊列，因此實際工作中都不建議這樣做的哦，阿里巴巴Java編程規範建議如下：

最後來個簡單的線程使用demo：

public class UseThreadPool{// 工作線程 static class Worker implements Runnable { private String taskName; private Random r = new Random(); public Worker(String taskName) { this.taskName = taskName; } public String getName() { return taskName; } @Override public void run() { System.out.println(Thread.currentThread().getName() + " 當前任務: " + taskName); try { TimeUnit.MILLISECONDS.sleep(r.nextInt(100) * 5); } catch (Exception e) { e.printStackTrace(); } } } static class CallWorker implements Callable<String> { private String taskName; private Random r = new Random(); public CallWorker(String taskName) { this.taskName = taskName; } public String getName() { return taskName; } @Override public String call() throws Exception { System.out.println(Thread.currentThread().getName() + " 當前任務 : " + taskName); return Thread.currentThread().getName() + ":" + r.nextInt(100) * 5; } } public static void main(String[] args) throws InterruptedException, ExecutionException { ExecutorService pool = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10), new ThreadPoolExecutor.DiscardOldestPolicy());// ExecutorService pool = Executors.newCachedThreadPool(); for (int i = 0; i < 5; i++) { Worker worker = new Worker("Runnable_" + i); pool.execute(worker); } for (int i = 0; i < 5; i++) { CallWorker callWorker = new CallWorker("CallWorker_" + i); Future<String> result = pool.submit(callWorker); System.out.println(result.get()); } pool.shutdown(); }}

ScheduledThreadPoolExecutor

周期性

執行任務的線程池，按照某種特定的計劃執行線程中的任務，有核心線程，但也有非核心線程，非核心線程的大小也為無限大。適用於執行周期性的任務。

看構造函數：調用的還是

構造函數，區別不同點在於任務隊列是用的DelayedWorkQueue，沒什麼新奇的了。

核心函數講解：

schedule只執行一次，任務還可以延時執行，傳入待執行任務跟延時時間。

scheduleAtFixedRate提交固定時間間隔的任務，提交任務，延時時間，已經循環時間間隔時間。這個的含義是只是在固定的時間間隔嘗試運行該任務。

scheduleWithFixedDelay提交固定延時間隔執行的任務。上一個任務執行完畢後等多久再執行下個任務，這個中間時間叫FixedDelay

其中scheduleAtFixedRate跟scheduleWithFixedDelay區別如下圖

scheduleAtFixedRate任務超時狀態，比如我們設定60s執行一次，其中第一個任務時長80s，第二個任務20s，第三個任務50s。第一個任務第0秒開始，第80s結束.第二個任務第80s開始，在第100s結束.第三個任務第120s秒開始，170s結束.第四個任務從180s開始.

簡單Mark個循環任務demo：

class ScheduleWorker implements Runnable {public final static int Normal = 0;//普通任務類型 public final static int HasException = -1;//會拋出異常的任務類型 public final static int ProcessException = 1;//拋出異常但會捕捉的任務類型 public static SimpleDateFormat formater = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); private int taskType; public ScheduleWorker(int taskType) { this.taskType = taskType; } @Override public void run() { if (taskType == HasException) { System.out.println(formater.format(new Date()) + " 異常產生"); throw new RuntimeException("有異常"); } else if (taskType == ProcessException) { try { System.out.println(formater.format(new Date()) + " 異常產生被捕捉"); throw new RuntimeException("異常被捕捉");//異常導致下個任務無法執行 } catch (Exception e) { System.out.println(" 異常被主播"); } } else { System.out.println("正常" + formater.format(new Date())); } }}public class SchTest{ public static void main(String[] args) { ScheduledThreadPoolExecutor schedule = new ScheduledThreadPoolExecutor(1); schedule.scheduleAtFixedRate(new ScheduleWorker(ScheduleWorker.HasException), 1000, 3000, TimeUnit.MILLISECONDS); // 任務在 1秒後執行 周期3秒 schedule.scheduleAtFixedRate(new ScheduleWorker(ScheduleWorker.Normal), 1000, 3000, TimeUnit.MILLISECONDS); }}

CompletionService

JDK8中新添加的一個類，攝像一個場景你去詢問兩個商品價格然後將價格保存資料庫。

ExecutorService executor =Executors.newFixedThreadPool(2);// 異步向電商 S1 詢價Future<Integer> f1 = executor.submit(()->getPriceByS1());// 異步向電商 S2 詢價Future<Integer> f2 = executor.submit(()->getPriceByS2());// 獲取電商 S1 報價並保存r=f1.get();executor.execute(()->save(r));// 獲取電商 S2 報價並保存r=f2.get();executor.execute(()->save(r));

上面的這個方案本身沒有太大問題，但是有個地方的處理需要你注意，那就是如果獲取電商 S1 報價的耗時很長，那麼即便獲取電商 S2 報價的耗時很短，也無法讓保存 S2 報價的操作先執行，因為這個主線程都阻塞在了 f1.get(),那我們如何解決了？解決方法：結果都存入到一個阻塞隊列中去。

// 創建阻塞隊列BlockingQueue<Integer> bq =new LinkedBlockingQueue<>();// 電商 S1 報價異步進入阻塞隊列 executor.execute(()->bq.put(f1.get()));// 電商 S2 報價異步進入阻塞隊列 executor.execute(()->bq.put(f2.get()));// 異步保存所有報價 for (int i=0; i<2; i++) {Integer r = bq.take(); executor.execute(()->save(r));}

在JDK8中不建議上面的工作都手動實現，JDK提供了

，它實現原理也是內部維護了一個阻塞隊列，它的核心功效就是讓先執行的任務先放到結果集。當任務執行結束就把任務的執行結果加入到阻塞隊列中，不同的是

是把任務執行結果的Future 對象加入到阻塞隊列中，而上面的示例代碼是把任務最終的執行結果放入了阻塞隊列中。

將

Executor

和

的功能融合在一起，

內部有個阻塞隊列。

接口的實現類是

ExecutorCompletionService

，這個實現類的構造方法有兩個，分別是：

ExecutorCompletionService(Executor executor)ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue)

這兩個構造方法都需要傳入一個線程池，如果不指定 completionQueue，那麼默認會使用無界的 LinkedBlockingQueue。任務執行結果的Future對象就是加入到 completionQueue 中。

// 創建線程池ExecutorService executor = Executors.newFixedThreadPool(2);// 創建 CompletionServiceCompletionService<Integer> cs = new ExecutorCompletionService<>(executor);// 異步向電商 S1 詢價cs.submit(()->getPriceByS1());// 異步向電商 S2 詢價cs.submit(()->getPriceByS2());// 將詢價結果異步保存到資料庫for (int i=0; i<2; i++) {Integer r = cs.take().get(); executor.execute(()->save(r));}

來一個整體的demo加深印象：

// 任務類class WorkTask implements Callable<Integer>{private String name; public WorkTask(String name) { this.name = name; } @Override public Integer call() { int sleepTime = new Random().nextInt(1000); try { Thread.sleep(sleepTime); } catch (InterruptedException e) { e.printStackTrace(); } return sleepTime; }}public class CompletionCase{ private final int POOL_SIZE = Runtime.getRuntime().availableProcessors(); private final int TOTAL_TASK = Runtime.getRuntime().availableProcessors(); public void selfByQueue() throws Exception { long start = System.currentTimeMillis(); // 統計所有任務休眠的總時長 AtomicInteger count = new AtomicInteger(0); ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE); // 創建線程池 BlockingQueue<Future<Integer>> queue = new LinkedBlockingQueue<Future<Integer>>();//容器存放提交給線程池的任務,list,map, for (int i = 0; i < TOTAL_TASK; i++) { Future<Integer> future = pool.submit(new WorkTask("要執行的第幾個任務" + i)); queue.add(future);//i=0 先進隊列，i=1的任務跟著進 } for (int i = 0; i < TOTAL_TASK; i++) { int sleptTime = queue.take().get(); // 檢查線程池任務執行結果 i=0先取到，i=1的後取到 System.out.println(" 休眠毫秒數 = " + sleptTime + " ms "); count.addAndGet(sleptTime); } pool.shutdown(); System.out.println("休眠時間" + count.get() + "ms,耗時時間" + (System.currentTimeMillis() - start) + " ms"); } public void testByCompletion() throws Exception { long start = System.currentTimeMillis(); AtomicInteger count = new AtomicInteger(0); // 創建線程池 ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE); CompletionService<Integer> cService = new ExecutorCompletionService<>(pool); // 向裡面扔任務 for (int i = 0; i < TOTAL_TASK; i++) { cService.submit(new WorkTask("執行任務" + i)); } // 檢查線程池任務執行結果 for (int i = 0; i < TOTAL_TASK; i++) { int sleptTime = cService.take().get(); System.out.println("休眠毫秒數 = " + sleptTime + " ms ..."); count.addAndGet(sleptTime); } pool.shutdown(); System.out.println("休眠時間 " + count.get() + "ms,耗時時間" + (System.currentTimeMillis() - start) + " ms"); } public static void main(String[] args) throws Exception { CompletionCase t = new CompletionCase(); t.selfByQueue(); t.testByCompletion(); }}