1.什么是線程池

線程池，通過創(chuàng)建了一定數(shù)量的線程并將其維護(hù)在一個池（即容器）中。當(dāng)有新的任務(wù)提交時，線程池會從池子中分配一個空閑線程來執(zhí)行任務(wù)，而不是每次都新建線程。執(zhí)行完任務(wù)后，線程不會被銷毀，而是返回到線程池中等待下一個任務(wù)。

2.線程池的優(yōu)勢

1. 避免線程的重復(fù)創(chuàng)建與銷毀：對于需要執(zhí)行的任務(wù)，如果每個任務(wù)都需要創(chuàng)建一個線程，線程執(zhí)行完任務(wù)后銷毀，那么會極大的造成資源的浪費(fèi)。一方面任務(wù)數(shù)量可能會很龐大，創(chuàng)建與之匹配的線程會對內(nèi)存造成嚴(yán)重消耗；另一方面，創(chuàng)建完的線程只工作一次，資本家看了落淚，md血虧啊。
2. 降低資源消耗：創(chuàng)建的線程反復(fù)利用，避免了創(chuàng)建與銷毀帶來的開銷
3. 提高工作的準(zhǔn)備時間：被提交的任務(wù)可以迅速被線程池中存儲的線程執(zhí)行，無需重新創(chuàng)建

3.原理

線程池中存在以下核心組件

• 線程池容器（存儲工作線程）
• 任務(wù)隊(duì)列（存儲需要執(zhí)行的任務(wù)）

下述代碼中，線程池使用HashSet存儲；任務(wù)隊(duì)列，使用的是這篇文章實(shí)現(xiàn)的BlockingQueue阻塞隊(duì)列

另外，單純的Thread線程能夠存儲的信息太少，因此我們創(chuàng)建Worker對象，extents Thread來包裝Thread

下圖是線程池的工作流程

大體來說，線程池執(zhí)行邏輯分為三大步驟

1. 如果current thread number < coreSize，創(chuàng)建核心線程執(zhí)行任務(wù)

   tip：

   • current thread number在源碼中，是有一個AtomicInteger變量ctl表示。ctl是核心線程池狀態(tài)控制器，它被分為兩個組成部分。其中，高三位表示runStatus，線程池狀態(tài)；低三位表示workCount，工作線程數(shù)量。
   • 選擇一個變量ctl同時存儲runStatus和workCount，可以通過一次CAS操作實(shí)現(xiàn)原子賦值，而不用兩次。

2. 如果核心線程創(chuàng)建失敗，或者核心線程數(shù)量過多，則將任務(wù)存儲在阻塞隊(duì)列中：在這一步中，存在非常多的細(xì)節(jié)。
   2.1. 如果當(dāng)前線程池不處于RUNNING狀態(tài)，嘗試創(chuàng)建救急線程運(yùn)行，不執(zhí)行入隊(duì)操作
   2.2. 如果入隊(duì)失敗，同樣創(chuàng)建救急線程
   2.3. 如果線程池處于運(yùn)行狀態(tài)，且入隊(duì)成功。進(jìn)行double-check，重新檢查線程池狀態(tài)ctl
   2.4. 如果此時線程池不處于RUNNIG狀態(tài)，移除剛?cè)腙?duì)的任務(wù)，并執(zhí)行reject策略
   2.5. 如果線程池依然處于RUNNIG狀態(tài)，且工作線程為0，創(chuàng)建救急線程，執(zhí)行任務(wù)
3. 如果上述步驟均失敗，創(chuàng)建救濟(jì)線程，如果依然失敗，執(zhí)行reject策略

4.代碼編寫

4.1 阻塞隊(duì)列

實(shí)現(xiàn)請看BlockingQueue阻塞隊(duì)列，本文不再贅述

4.2 ThreadPool線程池

/*** 線程池*/
@Slf4j
public class ThreadPool {// 核心線程數(shù)private int coreSize;// 阻塞隊(duì)列private BlockingQueue<Runnable> workQueue;// 隊(duì)列容量private int capacity;// 工作線程private final HashSet<Worker> workers = new HashSet<>();// todo: Worker（詳見下一部分）private final class Worker extents Thread { /*...*/ }public ThreadPool(int coreSize, int capacity) {this.coreSize = coreSize;this.capacity = capacity;this.workQueue = new BlockingQueue<>(capacity);}/*** 執(zhí)行task任務(wù). 如果當(dāng)前線程數(shù)量 < coreSize, 創(chuàng)建線程執(zhí)行* 否則加入阻塞隊(duì)列. 如果阻塞隊(duì)列已滿, 執(zhí)行當(dāng)前拒絕策略* @param task 需要執(zhí)行任務(wù)*/public void execute(Runnable task) {if (task == null)throw new NullPointerException("task is null");synchronized (workers) {if (workers.size() < coreSize) {// 創(chuàng)建線程執(zhí)行(我們傾向于創(chuàng)建新線程來執(zhí)行任務(wù), 而非已創(chuàng)建線程)log.info("創(chuàng)建worker");Worker worker = new Worker(task);workers.add(worker);worker.start(); // 千萬別寫成調(diào)用run方法, 否則主線程會阻塞(run不會開啟線程)}else {log.info("添加阻塞隊(duì)列");// 添加阻塞隊(duì)列workQueue.put(task);}}}
}

上述代碼實(shí)現(xiàn)簡易版線程池。

• workQueue：阻塞隊(duì)列，用于存儲待執(zhí)行的任務(wù)
• coreSize：核心線程數(shù)量
• capacity：阻塞隊(duì)列大小
• workers：工作線程的存儲容器（線程池），用HashSet實(shí)現(xiàn)。請注意，HashSet是線程不安全的，因此在對HashSet操作時，記得加鎖保證不會出現(xiàn)并發(fā)問題

本節(jié)對execute執(zhí)行邏輯進(jìn)行一定的簡化，暫時不考慮拒絕策略（后續(xù)介紹）。

• 如果當(dāng)前線程數(shù)量 < coreSize，創(chuàng)建核心線程并執(zhí)行任務(wù)
• 否則添加阻塞隊(duì)列

tip: 如果任務(wù)數(shù)量超過阻塞隊(duì)列容量，那么依據(jù)阻塞隊(duì)列的性質(zhì)，后續(xù)的所有線程都會阻塞，等待容量減少。

4.3 Worker工作線程

我們使用包裝過后的線程對象。且Worker是ThreadPool的內(nèi)部類

private final class Worker extends Thread {// 執(zhí)行的任務(wù)private Runnable task;Worker(Runnable task) {this.task = task;}/*** 執(zhí)行task任務(wù), 如果task為null, 則從workQueue工作隊(duì)列中獲取任務(wù)* 如果工作隊(duì)列中不存在等待執(zhí)行的任務(wù), 終止當(dāng)前Worker工作線程*/@Overridepublic void run() {while (task != null || (task = workQueue.take()) != null) {try {log.info("運(yùn)行任務(wù)");task.run();} catch (Exception e) {e.printStackTrace();} finally {task = null;}}// 移除當(dāng)前工作線程synchronized (workers) {workers.remove(this);}}
}

為了簡化代碼編寫，本文只存在核心線程。核心線程的工作是監(jiān)視阻塞隊(duì)列，獲取待執(zhí)行的任務(wù)并執(zhí)行

在run方法中，while循環(huán)的條件有二

• task != null: worker線程創(chuàng)建時，會分配第一個待執(zhí)行的任務(wù)。如果待執(zhí)行的任務(wù)不為null，則執(zhí)行任務(wù)
• task = workQueue.take()：worker線程持續(xù)監(jiān)視workQueue阻塞隊(duì)列中的任務(wù)，如果存在任務(wù)，獲取并執(zhí)行

tip: workQueue.take()是一個阻塞的方法，沒有時間的限制。也就是說，哪怕workQueue為空，該方法也會死等下去

4.4 代碼測試

import lombok.extern.slf4j.Slf4j;@Slf4j
public class Test {public static void main(String[] args) throws InterruptedException {ThreadPool threadPool = new ThreadPool(2, 5);for (int i = 0; i < 10; i++) {int j = i;// 任務(wù)創(chuàng)建時間為2s, 任務(wù)消費(fèi)時間顯著低于任務(wù)創(chuàng)建時間.// 因此本模型是個典型的快生產(chǎn), 慢消費(fèi)的模型threadPool.execute(() -> {try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}log.info(String.valueOf(j));});}}
}

控制臺輸出

21:07:34.189 [main] INFO com.fgbg.juc.ThreadPool - 創(chuàng)建worker
21:07:34.202 [main] INFO com.fgbg.juc.ThreadPool - 創(chuàng)建worker
21:07:34.202 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:07:34.205 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:07:34.205 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:07:34.205 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:07:34.205 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:07:34.205 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:07:34.203 [Thread-0] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:07:34.209 [Thread-1] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:07:36.223 [Thread-0] INFO com.fgbg.juc.Test - 0
21:07:36.223 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:07:36.223 [Thread-0] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:07:36.239 [Thread-1] INFO com.fgbg.juc.Test - 1
21:07:36.240 [Thread-1] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:07:36.240 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:07:38.239 [Thread-0] INFO com.fgbg.juc.Test - 2
21:07:38.239 [Thread-0] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:07:38.256 [Thread-1] INFO com.fgbg.juc.Test - 3
21:07:38.256 [Thread-1] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:07:40.250 [Thread-0] INFO com.fgbg.juc.Test - 4
21:07:40.250 [Thread-0] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:07:40.265 [Thread-1] INFO com.fgbg.juc.Test - 5
21:07:40.266 [Thread-1] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:07:42.252 [Thread-0] INFO com.fgbg.juc.Test - 6
21:07:42.252 [Thread-0] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:07:42.268 [Thread-1] INFO com.fgbg.juc.Test - 7
21:07:42.268 [Thread-1] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:07:44.260 [Thread-0] INFO com.fgbg.juc.Test - 8
21:07:44.275 [Thread-1] INFO com.fgbg.juc.Test - 9

線程池核心線程數(shù)為2，因此一開始迅速創(chuàng)建2個worker線程。但因?yàn)樽枞?duì)列容量為5，且每個線程工作需要2s，耗時遠(yuǎn)遠(yuǎn)小于任務(wù)產(chǎn)出的速度，因此隊(duì)列被迅速沾滿

當(dāng)提交第8個任務(wù)時，主線程進(jìn)入阻塞狀態(tài)，無法繼續(xù)提交任務(wù)（2個任務(wù)正在執(zhí)行 + 5個任務(wù)添加阻塞隊(duì)列 + 1個任務(wù)剛要入隊(duì)，就阻塞了）

當(dāng)?shù)谝粋€任務(wù)被執(zhí)行完成，Thread-0 Worker執(zhí)行阻塞隊(duì)列中的其他任務(wù)。此時存在多余位置，之前被阻塞主線程成功提交任務(wù)，并繼續(xù)循環(huán)

后續(xù)的流程大體一致，故不在做多余分析。

5. 拒絕策略

所謂拒絕策略，就是提供給調(diào)用方一個選擇。如果調(diào)用方提交了過量的任務(wù)，多余的任務(wù)作何種處理。

由上方代碼分析可知，我們一開始對于過量的任務(wù)，處理方案就是死等。但這種方案無法滿足其他特定的需求，比如某個場景對執(zhí)行速度有要求，等待一段時間后阻塞隊(duì)列依然無法處理額外的任務(wù)，那么主線程就要拋棄該任務(wù)。死等是處理的方式之一，但存在不少的局限性，我們需要更多的處理方式。

對于不同的處理方式，我們可以選擇將代碼寫死在ThreadPool中，但這樣太不靈活，對于不同的場景，我們需要添加大量if else。因此我們可以采用策略模式，將拒絕的行為抽象成一個接口，創(chuàng)建ThreadPool時，由調(diào)用方傳遞接口。這樣我們就可以在不改變ThreadPool內(nèi)部代碼的同時，改變ThreadPool面對超量任務(wù)的拒絕行為

5.1 抽象Reject接口

@FunctionalInterface
public interface RejectPolicy {// 執(zhí)行拒絕策略void reject(Runnable task, BlockingQueue<Runnable> workQueue);
}

5.2 BlockingQueue新增tryPut方法

tryPut方法，嘗試將元素立刻添加到阻塞隊(duì)列中，不支持阻塞等待

// 嘗試立即添加元素
public boolean tryPut(T task) {lock.lock();try {if (deque.size() == capacity) return false;deque.addLast(task);return true;} finally {lock.unlock();}
}

5.3 修改ThreadPool的execute方法

execute執(zhí)行task入隊(duì)操作時，如果入隊(duì)失敗（阻塞隊(duì)列已滿），則調(diào)用reject執(zhí)行拒絕策略

    public void execute(Runnable task) {if (task == null)throw new NullPointerException("task is null");synchronized (workers) {if (workers.size() < coreSize) {// 創(chuàng)建線程執(zhí)行(我們傾向于創(chuàng)建新線程來執(zhí)行任務(wù), 而非已創(chuàng)建線程)log.info("創(chuàng)建worker");Worker worker = new Worker(task);workers.add(worker);worker.start(); // 千萬別寫成調(diào)用run方法, 否則主線程會阻塞(run不會開啟線程)}else {log.info("添加阻塞隊(duì)列");/*----------------modify below-------------------*/// 添加阻塞隊(duì)列// workQueue.put(task);// 添加失敗if ( !workQueue.tryPut(task)) {// 執(zhí)行拒絕策略rejectPolicy.reject(task, workQueue);}}}}

5.4 ThreadPool線程池構(gòu)造函數(shù)修改

    // 拒絕策略private RejectPolicy rejectPolicy;public ThreadPool(int coreSize, int capacity, RejectPolicy rejectPolicy) {this(coreSize, capacity);this.rejectPolicy = rejectPolicy;}

5.5 拒絕策略實(shí)現(xiàn)

因?yàn)?code>RejectPolicy接口有@FunctionalInterface，支持lambda表達(dá)式，因此編寫的時候可以簡寫

1. 丟棄策略

(task, workQueue) -> {}

2. 移除最老元素

(task, workQueue) -> { workQueue.poll(); }

tip: 筆者自定義的BlockingQueue沒有實(shí)現(xiàn)poll方法，各位讀者如果感興趣，可以自行實(shí)現(xiàn)。需要注意的是，記得加鎖保證線程安全

3. 死等

(task, workQueue) -> { workQueue.put(task); }

4. 拋出異常

(task, workQueue) -> { new RuntimeException("workQueue is full"); }

5.6 代碼測試

@Slf4j
public class Test3 {public static void main(String[] args) throws InterruptedException {ThreadPool threadPool = new ThreadPool(2, 5, (task, workQueue) -> {log.info("任務(wù)丟棄");});for (int i = 0; i < 10; i++) {int j = i;threadPool.execute(() -> {try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}log.info(String.valueOf(j));});}}
}

上述代碼選擇的拒絕策略是丟棄

控制臺輸出

21:46:37.621 [main] INFO com.fgbg.juc.ThreadPool - 創(chuàng)建worker
21:46:37.630 [main] INFO com.fgbg.juc.ThreadPool - 創(chuàng)建worker
21:46:37.631 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:46:37.631 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:46:37.631 [Thread-0] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:46:37.631 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:46:37.631 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:46:37.631 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:46:37.631 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:46:37.631 [main] INFO com.fgbg.juc.Test3 - 任務(wù)丟棄
21:46:37.631 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:46:37.631 [main] INFO com.fgbg.juc.Test3 - 任務(wù)丟棄
21:46:37.632 [main] INFO com.fgbg.juc.ThreadPool - 添加阻塞隊(duì)列
21:46:37.632 [main] INFO com.fgbg.juc.Test3 - 任務(wù)丟棄
21:46:37.633 [Thread-1] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:46:39.636 [Thread-1] INFO com.fgbg.juc.Test3 - 1
21:46:39.636 [Thread-0] INFO com.fgbg.juc.Test3 - 0
21:46:39.636 [Thread-0] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:46:39.636 [Thread-1] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:46:41.644 [Thread-0] INFO com.fgbg.juc.Test3 - 3
21:46:41.645 [Thread-0] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:46:41.644 [Thread-1] INFO com.fgbg.juc.Test3 - 2
21:46:41.645 [Thread-1] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:46:43.651 [Thread-1] INFO com.fgbg.juc.Test3 - 5
21:46:43.651 [Thread-0] INFO com.fgbg.juc.Test3 - 4
21:46:43.651 [Thread-1] INFO com.fgbg.juc.ThreadPool - 運(yùn)行任務(wù)
21:46:45.657 [Thread-1] INFO com.fgbg.juc.Test3 - 6

由日志可知，第8，9，10號任務(wù)被丟棄。任務(wù)對應(yīng)的輸出為7，8，9。觀察輸出的數(shù)字，發(fā)現(xiàn)最大值為6。因此確認(rèn)了7~10號任務(wù)全部被拒絕，測試成功

6.全部代碼

BlockingQueue

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;// 消息隊(duì)列（阻塞隊(duì)列）
class BlockingQueue<T> {// 隊(duì)列private Deque<T> deque = new ArrayDeque<>();// 容量private int capacity;// 鎖private final ReentrantLock lock = new ReentrantLock();// 消費(fèi)者等待條件private Condition consumerWaitSet = lock.newCondition();// 生產(chǎn)者等待條件private Condition producerWaitSet = lock.newCondition();public BlockingQueue(int capacity) {this.capacity = capacity;}// 添加元素public void put(T element) {lock.lock();try {// 隊(duì)列已滿while (deque.size() == capacity) {try {// 阻塞等待producerWaitSet.await();} catch (InterruptedException e) {}}// 添加元素deque.addLast(element);// 喚醒其它線程consumerWaitSet.signal();} finally {lock.unlock();}}// 獲取元素public T take() {lock.lock();try {// 判空while (deque.size() == 0) {try {// 阻塞等待consumerWaitSet.await();} catch (InterruptedException e) {}}// 獲取元素T res = deque.pollFirst();producerWaitSet.signal();return res;} finally {lock.unlock();}}// 嘗試立即添加元素public boolean tryPut(T task) {lock.lock();try {if (deque.size() == capacity) return false;deque.addLast(task);return true;} finally {lock.unlock();}}
}

RejectPolicy

@FunctionalInterface
public interface RejectPolicy {// 執(zhí)行拒絕策略void reject(Runnable task, BlockingQueue<Runnable> workQueue);
}

ThreadPool

import lombok.extern.slf4j.Slf4j;import java.util.HashSet;/*** 線程池*/
@Slf4j
public class ThreadPool {// 核心線程數(shù)private int coreSize;// 阻塞隊(duì)列private BlockingQueue<Runnable> workQueue;// 隊(duì)列容量private int capacity;// 工作線程private final HashSet<Worker> workers = new HashSet<>();// 拒絕策略private RejectPolicy rejectPolicy;private final class Worker extends Thread {// 執(zhí)行的任務(wù)private Runnable task;Worker(Runnable task) {this.task = task;}/*** 執(zhí)行task任務(wù), 如果task為null, 則從workQueue工作隊(duì)列中獲取任務(wù)* 如果工作隊(duì)列中不存在等待執(zhí)行的任務(wù), 終止當(dāng)前Worker工作線程*/@Overridepublic void run() {while (task != null || (task = workQueue.take()) != null) {try {log.info("運(yùn)行任務(wù)");task.run();} catch (Exception e) {e.printStackTrace();} finally {task = null;}}// 移除當(dāng)前工作線程synchronized (workers) {workers.remove(this);}}}public ThreadPool(int coreSize, int capacity) {this.coreSize = coreSize;this.capacity = capacity;this.workQueue = new BlockingQueue<>(capacity);}public ThreadPool(int coreSize, int capacity, RejectPolicy rejectPolicy) {this(coreSize, capacity);this.rejectPolicy = rejectPolicy;}/*** 執(zhí)行task任務(wù). 如果當(dāng)前線程數(shù)量 < coreSize, 創(chuàng)建線程執(zhí)行* 否則加入阻塞隊(duì)列. 如果阻塞隊(duì)列已滿, 執(zhí)行當(dāng)前拒絕策略* @param task 需要執(zhí)行任務(wù)*/public void execute(Runnable task) {if (task == null)throw new NullPointerException("task is null");synchronized (workers) {if (workers.size() < coreSize) {// 創(chuàng)建線程執(zhí)行(我們傾向于創(chuàng)建新線程來執(zhí)行任務(wù), 而非已創(chuàng)建線程)log.info("創(chuàng)建worker");Worker worker = new Worker(task);workers.add(worker);worker.start(); // 千萬別寫成調(diào)用run方法, 否則主線程會阻塞(run不會開啟線程)}else {log.info("添加阻塞隊(duì)列");// 添加阻塞隊(duì)列// workQueue.put(task);// 添加失敗if ( !workQueue.tryPut(task)) {// 執(zhí)行拒絕策略rejectPolicy.reject(task, workQueue);}}}}
}

Test

import lombok.extern.slf4j.Slf4j;@Slf4j
public class Test {public static void main(String[] args) throws InterruptedException {ThreadPool threadPool = new ThreadPool(2, 5);for (int i = 0; i < 10; i++) {int j = i;threadPool.execute(() -> {try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}log.info(String.valueOf(j));});}}
}

Test3

@Slf4j
public class Test3 {public static void main(String[] args) throws InterruptedException {ThreadPool threadPool = new ThreadPool(2, 5, (task, workQueue) -> {log.info("任務(wù)丟棄");});for (int i = 0; i < 10; i++) {int j = i;threadPool.execute(() -> {try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}log.info(String.valueOf(j));});}}
}