手動造一個執行緒池(Java)
本次自己實現一個簡單的執行緒池，主要是為了後續看 ThreadPool 的原始碼做準備的，是從別人的程式碼中改進的，從看別人的原始碼中學到一些東西，所以特意把這篇文章寫出來，方便以後自己去回顧自己是如何學習。當然也希望分享出來可以對別人產生良好的影響！
使用Java的執行緒池
在自己實現一個執行緒池之前，首先要知道怎麼用。因為知道怎麼用之後才能去理解一些程式碼的編寫。關於怎麼用這裡就不再多加贅述了，百度或者谷歌一下就好，為了不讓讀者花過多的時間去找，我找了一篇文章，說得比較清楚。
總覽
我們可以看到，除了 Thread 和 Runnable ，其他都是我們自己定義的，下面我們來逐一說明。
在我們開始分析之前，先說下執行緒池的工作流程，也方便大家後面看的時候心理有一個底。
執行緒池顧名思義就是一個存放多個執行緒的池子。那麼在計算機語言中，我們就是用資料結構來存放執行緒，在本執行緒池中用的是一個佇列來存放要處理任務的執行緒。所以線上程池一啟動，執行緒池裡面就應該有一定數量的執行緒數目了，那麼這個執行緒的數目是多少我們先不用管，只需要知道有一些執行緒在等待使用者把所需要執行緒執行的任務放進池子裡面。然後執行緒池裡面的執行緒就會自動幫你執行任務啦。
當然有些人說，我執行一個任務就建立一個執行緒就好了呀，何必大費周章呢。我們需要知道，來一個任務就建立一個執行緒，
-
建立執行緒需要時間 ，影響響應速度。
-
系統資源有限，如果有數以萬計的執行緒需要建立，會大大消耗系統資源，會降低系統的穩定性。
其實有很多工的時候，有些執行緒只是處理一些很輕的任務，很快就完成了，那麼如果下一個任務剛好到達的時候，之前的執行緒也剛好完成工作了，那麼這個執行緒就順便接下到來的任務，這樣的話豈不是提高了響應速度，然後又重複利用了執行緒，降低系統資源的損耗。豈不是一舉兩得。
之前都是恰巧，那麼我們稍微放寬一點條件。如果執行緒執行完任務了，就先別退出唄。而是在等待執行任務，這個執行緒就可以看做被賦予 執行任務 的命令！**就等著任務來，任務一來，我就去執行。任務執行結束，執行緒就等，直到下一個任務來。周而復始，直到手動關閉！**這就是執行緒池的本質。
那麼問題來了，執行緒池裡面只有5個執行緒在等待執行任務，可是同時來了10個任務需要執行，那麼有5個任務被執行了，剩下那5個放哪裡？難道被丟棄？這可不是我們設計執行緒池的初衷！你肯定可以想到，肯定是拿一樣資料結構去儲存剩下的執行緒呀！(我們用佇列儲存，然後稱為工作佇列。)因為執行緒處理任務的時間是不一定的，肯定是有些執行緒處理的快，有些慢。所以誰先處理的快，誰就去處理剩下的任務。正所謂能者多勞！
再丟擲一個問題，假如前面5個執行緒執行得很慢，那麼後面那5個執行緒就需要等很久，這時候還不如直接建立執行緒去操作呢，沒錯，執行緒池在設計的時候也想到過這個問題，關於這個問題在後面我們設計的時候會說道，這裡就先往下看吧！
既然涉及到多執行緒，那麼肯定就涉及到同步的問題，對哪個物件需要同步呢？當然是任務佇列啦。我們需要知道很有可能同時會有很多個執行緒對 同一個任務佇列 取任務和放任務的，所以為了實現同步，我們這裡用了 synchronized 關鍵字實現同步,也就是對這個任務佇列加一把鎖，哪個執行緒可以拿到操作任務佇列的鎖哪個執行緒就可以領取任務。沒拿到這把鎖的執行緒就死等，除非被中斷或者手動關閉。
這裡需要注意的是 掛起阻塞 和 等待拿鎖 的區別。
-
掛起阻塞是該執行緒拿到鎖之後呼叫
await方法才會進入的狀態，前提是先拿到鎖。被通知之後就會被喚醒，然後從await之後的程式碼執行。 -
等待拿鎖是別的執行緒還在佔有鎖，此時的執行緒還沒拿到鎖，就會進入這個鎖的entrySet序列等待，直到鎖被釋放然後再去搶，搶到為止！
經過上面的講解，我們可以基本瞭解了執行緒池的設計思想和原理，下面補充點內容。
-
執行緒池內部有兩個資料結構（佇列）分別存放需要 執行任務的執行緒 (也叫工作執行緒)和所需要被**執行的任務*。
-
執行緒池初始化的執行緒放在 工作佇列 裡面，使用者想要執行的任務放在 任務佇列
-
在使用者新增任務之後，會通知工作佇列的執行緒去取任務啦！
-
工作佇列的執行緒如果有空並且任務佇列不為空，哪個執行緒拿到鎖哪個執行緒就可以在任務佇列取任務，然後任務佇列的任務數就-1。
-
很多個執行緒去拿鎖的時候，只能有一個執行緒拿到。 其他沒拿到鎖的執行緒不是阻塞等待，而是等待拿鎖！
-
如果拿到鎖之後任務佇列為空，就掛起阻塞。如果被通知喚醒，繼續執行3 4 5 6操作。
先看看我們這個整個執行緒池的流程圖，這樣設計的時候就知道怎麼回事了！
過程
BaseThreadPool
先看看這個類的基本屬性
public class BaseThreadPool extends Thread implements ThreadPool {
/*初始化執行緒數*/
private int initSize;
/*最大工作執行緒數*/
private int maxSize;
/*核心執行緒數*/
private int coreSize;
/*當前活躍執行緒數*/
privateint activityCount = 0;
/*指定任務佇列的大小數*/
private int queueSize;
/*建立工作執行緒的工廠，在構造方法由執行緒池規定好*/
private ThreadFactory threadFactory;
/*1. 任務佇列，在構造方法由執行緒池規定好*/
private RunnableQueue runnableQueue;
//2. 工作佇列
private final static Queue<ThreadTask> threadQueue = new ArrayDeque<>();
//3. 本執行緒池預設的拒絕策略
private final static DenyPolicy DEFAULT_DENY_POLICY = new DenyPolicy.IgnoreDenyPolicy();
/*4. 預設的執行緒工廠*/
private final static ThreadFactory DEFAULT_THREAD_FACTORY =new DefaultThreadFactory();
/*執行緒池是否關閉，預設為false*/
boolean isShutdown = false;
privatelong keepAliveTime;
privateTimeUnit timeUnit ;
複製程式碼
由上面的屬性我們知道，我們自定義的執行緒池這個類是依賴於幾個類的。
依次是 RunnableQueue ， DenyPolicy ， ThreadFactory 。
並且由總覽圖我們知道， BaseThreadPool 是實現了我們定義的 ThreadPool 介面和繼承了Thread類，並且重寫了run方法
run 裡面的邏輯到後面再分析,這裡可以先跳過這裡。
@Override
public void run() { // BaseThreadPool
while (!isShutdown && !isInterrupted()){
try {
timeUnit.sleep(keepAliveTime);
} catch (InterruptedException e) {
//到這裡就是關閉執行緒池了
isShutdown = true;
continue;
}
//這裡同步程式碼塊，保證了每次訪問的時候都是最新的資料！
synchronized (this){
if(isShutdown) break;
//任務佇列不為空，並且當前可以工作的執行緒小於coreCount，那麼說明工作執行緒數不夠，就先增加到maxSize
//比如說coreSize 為20，initSize為10，maxSize 為30,
//突然一下子來了20分執行緒進來，但是工作執行緒只有15個，由於某種原因可能那15個工作現場還沒執行完，那麼此時的任務佇列肯定還有剩餘的，發現此時執行緒還沒到coreSize
//那麼就直接開maxSize個執行緒先把
if(runnableQueue.size() > 0){
for (int i = runnableQueue.size(); i < maxSize; i++) {
newThread();
}
}
//任務佇列為空，並且當前可以工作的執行緒數大於coreCount，工作執行緒數太多啦！那麼就減少到coreCount
if(runnableQueue.size() == 0 &&activityCount > coreSize){
for (int i = coreSize; i < activityCount; i++) {
removeThread();
}
}
}
}
}
複製程式碼
我們先來看下BaseThreadPool的構造方法
//1 使用者傳入初始化執行緒數，最大執行緒數，核心執行緒數，和任務佇列的大小即可
public BaseThreadPool(int initSize, int maxSize, int coreSize,int queueSize) {
/*這裡建立執行緒的工廠和拒絕策略都是用自己定義好的物件*/this(initSize,maxSize,coreSize,queueSize,DEFAULT_THREAD_FACTORY,DEFAULT_DENY_POLICY,10,TimeUnit.SECONDS);
}
// 2
public BaseThreadPool(int initSize, int maxSize, int coreSize, int queueSize, ThreadFactory threadFactory, DenyPolicy denyPolicy, long keepAliveTime, TimeUnit timeUnit) {
this.initSize = initSize; //初始化執行緒池的初始化執行緒數
this.maxSize = maxSize; // 初始化執行緒池可以擁有最大的執行緒數
this.coreSize = coreSize; // 這個值的意義後面說
this.threadFactory = threadFactory; //初始化建立執行緒池的工廠
//自定義存放任務的佇列
this.runnableQueue = new LinkRunnableQueue(queueSize,denyPolicy,this); //RunnableQueue的實現類，自己定義
this.keepAliveTime = keepAliveTime;
this.timeUnit = timeUnit;
this.init(); //初始化函式
}
// ---init()
public void init(){
/*啟動本執行緒池*/
this.start();//BaseThreadPool 繼承了 Thread,原因後面說
/*初始化initSize個執行緒線上程池中*/
for (int i = 0; i < initSize; i++) {
newThread();
}
}
//newThread()
public void newThread(){
/*建立工作執行緒，然後讓工作執行緒等待任務到來被喚醒*/
Woker woker = new Woker(runnableQueue);
Thread thread = threadFactory.createThread(woker);
/*將執行緒和任務包裝在一起*/
ThreadTask threadTask = new ThreadTask(thread,woker);
threadQueue.offer(threadTask);
this.activityCount++;
/*啟動剛才新建的執行緒*/
thread.start();
}
// 再看看DefaultThreadFactory，就是
/*工廠建立一個新的執行緒*/
public class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger GROUP_COUNTER= new AtomicInteger(0); //執行緒組號
//計數
private staticAtomicInteger COUNTER = new AtomicInteger(1);
private static final ThreadGroup group= new ThreadGroup("MyThreadPool-" + GROUP_COUNTER.getAndIncrement());
@Override
public Thread createThread(Runnable runnable) {
return new Thread(group,runnable,"threadPool-" + COUNTER.getAndIncrement());
}
}
複製程式碼
這裡說明一下，我們是可以這樣 new Thread(new Runnable(){....}).start 建立並且啟動執行緒的。就是呼叫 Thread 需要傳入一個 Runnable 例項的建構函式例項化 Thread 類，通過重寫 Runnable 裡面的 run 方法就可以指定執行緒在啟動的時候需要做的事。
我們看到 DefaultThreadFactory 就只有一個建立執行緒的方法，就是把執行緒啟動後需要做的任務指定一下和重新命名一下執行緒，就是用上面說明的方法。所以傳給需要傳給 createThread 方法一個實現 Runnable 的類。而這個類就是 Woker
我們看下Woker的程式碼
//------------Woker BaseThreadPool依賴的類
/*工作執行緒的任務*/
public class Woker implements Runnable{
/*任務佇列，方便後面取出任務*/
private RunnableQueue runnableQueue;
/*方便判斷該內部任務對應的執行緒是否執行，確保可見性!*/
private volatile boolean running = true;
public Woker(RunnableQueue runnableQueue) {
this.runnableQueue = runnableQueue;
}
@Override
public void run() {
/*當前對應的執行緒正在執行並且沒有被中斷*/
while (running && !Thread.currentThread().isInterrupted()){
//呼叫take的時候，如果任務佇列沒任務就會阻塞在這，直到拿到任務
Runnable task = runnableQueue.take();
task.run();
}
}
public void stop(){
running = false;
}
}
複製程式碼
我們看到 run 方法，這個任務就是去到 任務佇列 裡面取任務，然後執行。直到當前工作停止或者當前執行緒被中斷。而這個 任務佇列 就是我們在呼叫建構函式的時候指定的物件，也就是這段程式碼
this.runnableQueue = new LinkRunnableQueue(queueSize,denyPolicy,this);
接下來看下 LinkRunnableQueue 是怎麼實現的
public class LinkRunnableQueue implements RunnableQueue{//BaseThreadPool依賴的類
//指定任務佇列的大小
private int limit;
//也是使用BaseThreadPool傳進來的預設拒絕策略
private DenyPolicy denyPolicy;
//這裡傳進BaseThreadPool例項
private ThreadPool threadPool;
//這個就是真正儲存Runnable例項物件的資料結構！也就是一個連結串列
private LinkedList<Runnable> queue = new LinkedList<>();
//建構函式，也就是初始化這個類的屬性
public LinkRunnableQueue(int queueSize,DenyPolicy denyPolicy,ThreadPool pool) {
this.limit = queueSize;
this.denyPolicy = denyPolicy;
this.threadPool = pool;
}
//任務佇列新增任務，這個方法一般由執行緒池的execute方法呼叫
@Override
public void offer(Runnable runnable) {
//因為任務佇列只有一個，可能會有多個執行緒同時操作任務佇列，所以要考慮同步問題
//取得queue的鎖才能加入任務，拿不到所就進入queue的entrySet
synchronized (queue){
if(queue.size() > limit){
//如果此時任務佇列超過限制的值，那麼就拒絕！
denyPolicy.reject(runnable,threadPool);
}else{
//把任務加入到任務佇列唄
queue.addLast(runnable);
//喚醒等待的執行緒，這些執行緒在queue的waitSet裡面，要結合take方法
queue.notifyAll();
}
}
}
//執行緒從任務佇列裡面拿任務，如果拿不到就會阻塞，直到有任務來並且搶到
@Override
public Runnable take() {
//這裡之前也說過了，要先拿到鎖才能拿任務
synchronized (queue){
//如果任務佇列為空，那麼肯定拿不了，所以就等待唄
while (queue.size() == 0){
try {
//這個執行緒在這裡就等待讓出鎖，直到執行offer方法從而被喚醒，然後
//再重新搶到鎖，這裡是個迴圈，如果被喚醒後，也搶到鎖了，但是佇列
//還是空的話，繼續等待
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//到這裡執行這個方法的執行緒就是搶到鎖了，然後得到任務啦！
return queue.removeFirst();
}
}
//返回呼叫該方法時任務佇列有多少個任務在等待
@Override
public int size() {
synchronized (queue){
return queue.size();
}
}
}
複製程式碼
程式碼的註釋已經解釋得很清楚了，這裡主要是瞭解為什麼Work中的 Runnable task = runnableQueue.take() 中沒有任務會阻塞等待，本質就是
1 拿到queue物件鎖之後，任務佇列沒任務，釋放掉真正儲存任務的物件的物件鎖，從而進入該物件的waitSet佇列裡面等待被喚醒。
2 當然如果沒拿到鎖也會一直等待拿到鎖，然後像1一樣.
如果看到這裡看不太明白的，大家可以先回去看一下java執行緒的基本知識和 synchronized 的詳解，這樣可以更好地把知識串聯起來！
接下來我們再看下 工作佇列 是什麼樣子。
ThreadTask在BaseThreadPool的一個內部類
//把工作執行緒和內部任務繫結在一起
class ThreadTask{
Thread thread;
Woker woker;
public ThreadTask(Thread thread, Woker woker) {
this.thread = thread;
this.woker = woker;
}
}
複製程式碼
從上面的程式碼我們知道，ThreadTask就是把一個工作執行緒和一個工作執行緒的任務封裝在一起而已，這裡主要是為了後面 執行緒池關閉 的時候可以讓執行緒需要做的任務停止！
執行緒池關閉的操作 ， BaseThreadPool 類的方法
/*shutdown 就要把 Woker 給停止 和 對應的執行緒給中斷*/
@Override
public void shutDown() {
synchronized (this){
if(isShutDown())
return;
//設定標誌位，讓執行緒池執行緒也執行完run方法，然後退出執行緒。
isShutdown = true;
/*全部執行緒停止工作*/
for (ThreadTask task: threadQueue
) {
//1 這裡就是把Woker例項物件的running置為false
task.woker.stop();
//2 中斷執行對應任務的執行緒
task.thread.interrupt();
}
}
}
複製程式碼
可以看到關閉執行緒池，就是遍歷存放工作執行緒的佇列， 1和2都是破壞Woker物件的while迴圈條件 ，從而讓Woker物件的 run 方法執行結束。（這裡大家可以看下Woker這個類的 run 方法就明白我說的了）
我們在開始的時候說過， BaseThreadPool 啟動的時候其實也是一個執行緒，在它的 init 方法中就呼叫了 start 方法表示執行 run 裡面的邏輯，之前我們看了run的程式碼，但是沒分析，現在就來分析吧
@Override
public void run() { //BaseThreadPool類的方法
//還記得shutDown（）方法裡面的 isShutdown = true語句嗎?
//作用就是為了讓這裡下一次判斷while迴圈的時候退出，然後執行完run啦！
while (!isShutdown && !isInterrupted()){
try {
timeUnit.sleep(keepAliveTime);
} catch (InterruptedException e) {
//如果執行緒池這個執行緒被中斷
//到這裡就是關閉執行緒池了,也是把isShutdown設定為我true！
isShutdown = true;
continue;
}
//這裡同步程式碼塊，保證了每次訪問的時候都是最新的資料！
synchronized (this){
if(isShutdown) break;
//任務佇列不為空，並且當前可以工作的執行緒小於coreCount，那麼說明工作//執行緒數不夠，就先增加到maxSize.
//比如說coreSize 為20，initSize為10，maxSize 為30,
//突然一下子來了20分執行緒進來，但是工作執行緒只有15個，由於某種原因可能
//那15個工作現場還沒執行完，那麼此時的任務佇列肯定還有剩餘的，發現此
//時執行緒還沒到coreSize
//那麼就直接開maxSize個執行緒先把
//如果發現現在工作的的執行緒已經過了coreSize就先不增加執行緒數啦
if(runnableQueue.size() > 0 && activityCount < coreSize){
for (int i = runnableQueue.size(); i < maxSize; i++) {
newThread();
}
}
//任務佇列為空，並且當前可以工作的執行緒數大於coreCount，工作執行緒數太多啦！那麼就減少到coreCount基本大小把
if(runnableQueue.size() == 0 &&activityCount > coreSize){
for (int i = coreSize; i < activityCount; i++) {
removeThread();
}
}
}
}
}
//----------removeThread()
//執行緒池中去掉某個工作執行緒,這裡的操作是不是很類似shutDown的內容
public void removeThread(){
this.activityCount--;
ThreadTask task = threadQueue.remove();
task.woker.stop();//就是破壞Woker物件的while迴圈的條件
}
複製程式碼
上面的註釋講解的比較清楚，有啥不懂的多看幾篇，自己模擬一下思路就好啦!
在 run 方法中，重要的是關於執行緒池中的執行緒數量的動態變化的部分。
coreSize:執行緒池基本的大小，相當於一個分界線
initSize:執行緒池的初始化大小，這枚啥好說的
activityCount：當前工作執行緒的數量
maxSIze:執行緒池中最大的執行緒數目
說一下它們之間的關係
任務佇列不為空的情況下
-
activityCount < coreSize的時候，就說明執行緒池的數量沒到達基本大小，就新增執行緒，直接新增到最大！
-
activityCount >= coreSize的時候，說明當前執行緒池的工作執行緒數量已經到達基本大小，有任務來就需要等一下啦！
注意：這裡的擴容機制只是簡單地擴容，Java中實現的執行緒池並不是像我說那樣擴容的，這就解決了開頭的問題啦，具體的到時候還是分析原始碼的時候再說把！這裡只是簡單地實現一下！
測試
測試程式碼
package blogDemo.ThreadDemo;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class Test {
public static void main(String[] args) {
ThreadPool threadPool = new BaseThreadPool(4,30,6,30);
for (int i = 0; i < 20; i++) {
threadPool.execute(() -> {
System.out.println(Thread.currentThread().getName() + " is running and done.");
});
}
}
}
複製程式碼
測試結果
總結
本片文章就寫到這裡啦，大家看文章的時候可以一邊看原始碼一邊看解釋，這樣會更加容易理解，希望對讀者後面理解java自帶執行緒池有所幫助，下一篇文章就分析java自帶的執行緒池的原始碼啦！