執行緒池：從設計思想到原始碼決議

前言
初識執行緒池
- 執行緒池優勢
- 執行緒池設計思路
深入執行緒池
- 構造方法
- 任務佇列
- 拒絕策略
- 執行緒池狀態
- 初始化&容量調整&關閉
使用執行緒池
- ThreadPoolExecutor
- Executors封裝執行緒池
解讀執行緒池
- execute()
- addWorker()
- Worker類
- runWorker()
- processWorkerExit()

前言

各位小伙伴兒，春節已經結束了，在此獻上一篇肝了一個春節假期的遲來的拜年之作，希望讀者朋友們都能有識訓，

根據穆氏哲學，投入越多，識訓越大，我作此文時，披肝瀝膽，汝讀此文時，一目十行，我們的識訓當然不同，
…
那怎么有更大的識訓呢？根據科學研究，當你為一個事物付出時（包括情緒付出），你就會對它更專注，最直接的付出是什么呢？當然是點贊和收藏啦，

初識執行緒池

我們知道，執行緒的創建和銷毀都需要映射到作業系統，因此其代價是比較高昂的，出于避免頻繁創建、銷毀執行緒以及方便執行緒管理的需要，執行緒池應運而生，

執行緒池優勢

降低資源消耗：執行緒池通常會維護一些執行緒（數量為 corePoolSize），這些執行緒被重復使用來執行不同的任務，任務完成后不會銷毀，在待處理任務量很大的時候，通過對執行緒資源的復用，避免了執行緒的頻繁創建與銷毀，從而降低了系統資源消耗，
提高回應速度：由于執行緒池維護了一批 alive 狀態的執行緒，當任務到達時，不需要再創建執行緒，而是直接由這些執行緒去執行任務，從而減少了任務的等待時間，
提高執行緒的可管理性：使用執行緒池可以對執行緒進行統一的分配，調優和監控，

執行緒池設計思路

有句話叫做藝術來源于生活，編程語言也是如此，很多設計思想能映射到日常生活中，比如面向物件思想、封裝、繼承，等等，今天我們要說的執行緒池，它同樣可以在現實世界找到對應的物體——工廠，

先假想一個工廠的生產流程：
在這里插入圖片描述

工廠中有固定的一批工人，稱為正式工人，工廠接收的訂單由這些工人去完成，當訂單增加，正式工人已經忙不過來了，工廠會將生產原料暫時堆積在倉庫中，等有空閑的工人時再處理（因為工人空閑了也不會主動處理倉庫中的生產任務，所以需要調度員實時調度），倉庫堆積滿了后，訂單還在增加怎么辦？工廠只能臨時擴招一批工人來應對生產高峰，而這批工人高峰結束后是要清退的，所以稱為臨時工，當時臨時工也以招滿后（受限于工位限制，臨時工數量有上限），后面的訂單只能忍痛拒絕了，

我們做如下一番映射：

工廠——執行緒池
訂單——任務（Runnable）
正式工人——核心執行緒
臨時工——普通執行緒
倉庫——任務佇列
調度員——getTask()

getTask()是一個方法，將任務佇列中的任務調度給空閑執行緒，在解讀執行緒池有詳細介紹

映射后，形成執行緒池流程圖如下，兩者是不是有異曲同工之妙？
在這里插入圖片描述
這樣，執行緒池的作業原理或者說流程就很好理解了，提煉成一個簡圖：

深入執行緒池

那么接下來，問題來了，執行緒池是具體如何實作這套作業機制的呢？從Java執行緒池Executor框架體系可以看出：執行緒池的真正實作類是ThreadPoolExecutor，因此我們接下來重點研究這個類，
圖片來自網路

構造方法

研究一個類，先從它的構造方法開始，ThreadPoolExecutor提供了4個有參構造方法：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          RejectedExecutionHandler handler) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

解釋一下構造方法中涉及到的引數：

corePoolSize（必需）：核心執行緒數，即池中一直保持存活的執行緒數，即使這些執行緒處于空閑，但是將allowCoreThreadTimeOut引數設定為true后，核心執行緒處于空閑一段時間以上，也會被回收，
maximumPoolSize（必需）：池中允許的最大執行緒數，當核心執行緒全部繁忙且任務佇列打滿之后，執行緒池會臨時追加執行緒，直到總執行緒數達到maximumPoolSize這個上限，
keepAliveTime（必需）：執行緒空閑超時時間，當非核心執行緒處于空閑狀態的時間超過這個時間后，該執行緒將被回收，將allowCoreThreadTimeOut引數設定為true后，核心執行緒也會被回收，
unit（必需）：keepAliveTime引數的時間單位，有：TimeUnit.DAYS（天）、TimeUnit.HOURS（小時）、TimeUnit.MINUTES（分鐘）、TimeUnit.SECONDS（秒）、TimeUnit.MILLISECONDS（毫秒）、TimeUnit.MICROSECONDS（微秒）、TimeUnit.NANOSECONDS（納秒）
workQueue（必需）：任務佇列，采用阻塞佇列實作，當核心執行緒全部繁忙時，后續由execute方法提交的Runnable將存放在任務佇列中，等待被執行緒處理，
threadFactory（可選）：執行緒工廠，指定執行緒池創建執行緒的方式，
handler（可選）：拒絕策略，當執行緒池中執行緒數達到maximumPoolSize且workQueue打滿時，后續提交的任務將被拒絕，handler可以指定用什么方式拒絕任務，

放到一起再看一下：
在這里插入圖片描述

任務佇列

使用ThreadPoolExecutor需要指定一個實作了BlockingQueue介面的任務等待佇列，在ThreadPoolExecutor執行緒池的API檔案中，一共推薦了三種等待佇列，它們是：SynchronousQueue、LinkedBlockingQueue和ArrayBlockingQueue；

SynchronousQueue：同步佇列，這是一個內部沒有任何容量的阻塞佇列，任何一次插入操作的元素都要等待相對的洗掉/讀取操作，否則進行插入操作的執行緒就要一直等待，反之亦然，
LinkedBlockingQueue：無界佇列（嚴格來說并非無界，上限是Integer.MAX_VALUE），基于鏈表結構，使用無界佇列后，當核心執行緒都繁忙時，后續任務可以無限加入佇列，因此執行緒池中執行緒數不會超過核心執行緒數，這種佇列可以提高執行緒池吞吐量，但代價是犧牲記憶體空間，甚至會導致記憶體溢位，另外，使用它時可以指定容量，這樣它也就是一種有界佇列了，
ArrayBlockingQueue：有界佇列，基于陣列實作，在執行緒池初始化時，指定佇列的容量，后續無法再調整，這種有界佇列有利于防止資源耗盡，但可能更難調整和控制，

另外，Java還提供了另外4種佇列：

PriorityBlockingQueue：支持優先級排序的無界阻塞佇列，存放在PriorityBlockingQueue中的元素必須實作Comparable介面，這樣才能通過實作compareTo()方法進行排序，優先級最高的元素將始終排在佇列的頭部；PriorityBlockingQueue不會保證優先級一樣的元素的排序，也不保證當前佇列中除了優先級最高的元素以外的元素，隨時處于正確排序的位置，
DelayQueue：延遲佇列，基于二叉堆實作，同時具備：無界佇列、阻塞佇列、優先佇列的特征，DelayQueue延遲佇列中存放的物件，必須是實作Delayed介面的類物件，通過執行時延從佇列中提取任務，時間沒到任務取不出來，更多內容請見DelayQueue，
LinkedBlockingDeque：雙端佇列，基于鏈表實作，既可以從尾部插入/取出元素，還可以從頭部插入元素/取出元素，
LinkedTransferQueue：由鏈表結構組成的無界阻塞佇列，這個佇列比較特別的時，采用一種預占模式，意思就是消費者執行緒取元素時，如果佇列不為空，則直接取走資料，若佇列為空，那就生成一個節點（節點元素為null）入隊，然后消費者執行緒被等待在這個節點上，后面生產者執行緒入隊時發現有一個元素為null的節點，生產者執行緒就不入隊了，直接就將元素填充到該節點，并喚醒該節點等待的執行緒，被喚醒的消費者執行緒取走元素，

拒絕策略

執行緒池有一個重要的機制：拒絕策略，當執行緒池workQueue已滿且無法再創建新執行緒池時，就要拒絕后續任務了，拒絕策略需要實作RejectedExecutionHandler介面，不過Executors框架已經為我們實作了4種拒絕策略：

AbortPolicy（默認）：丟棄任務并拋出RejectedExecutionException例外，
CallerRunsPolicy：直接運行這個任務的run方法，但并非是由執行緒池的執行緒處理，而是交由任務的呼叫執行緒處理，
DiscardPolicy：直接丟棄任務，不拋出任何例外，
DiscardOldestPolicy：將當前處于等待佇列列頭的等待任務強行取出，然后再試圖將當前被拒絕的任務提交到執行緒池執行，

執行緒工廠指定創建執行緒的方式，這個引數不是必選項，Executors類已經為我們非常貼心地提供了一個默認的執行緒工廠：

/**
 * The default thread factory
 */
static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

執行緒池狀態

執行緒池有5種狀態：

volatile int runState;
// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

runState表示當前執行緒池的狀態，它是一個 volatile 變數用來保證執行緒之間的可見性，

下面的幾個static final變數表示runState可能的幾個取值，有以下幾個狀態：

RUNNING：當創建執行緒池后，初始時，執行緒池處于RUNNING狀態；
SHUTDOWN：如果呼叫了shutdown()方法，則執行緒池處于SHUTDOWN狀態，此時執行緒池不能夠接受新的任務，它會等待所有任務執行完畢；
STOP：如果呼叫了shutdownNow()方法，則執行緒池處于STOP狀態，此時執行緒池不能接受新的任務，并且會去嘗試終止正在執行的任務；
TERMINATED：當執行緒池處于SHUTDOWN或STOP狀態，并且所有作業執行緒已經銷毀，任務快取佇列已經清慷訓執行結束后，執行緒池被設定為TERMINATED狀態，

初始化&容量調整&關閉

1、執行緒初始化

默認情況下，創建執行緒池之后，執行緒池中是沒有執行緒的，需要提交任務之后才會創建執行緒，

在實際中如果需要執行緒池創建之后立即創建執行緒，可以通過以下兩個方法辦到：

prestartCoreThread()：boolean prestartCoreThread()，初始化一個核心執行緒
prestartAllCoreThreads()：int prestartAllCoreThreads()，初始化所有核心執行緒，并回傳初始化的執行緒數

public boolean prestartCoreThread() {
    return addIfUnderCorePoolSize(null); //注意傳進去的引數是null
}

public int prestartAllCoreThreads() {
    int n = 0;
    while (addIfUnderCorePoolSize(null))//注意傳進去的引數是null
        ++n;
    return n;
}

2、執行緒池關閉

ThreadPoolExecutor提供了兩個方法，用于執行緒池的關閉：

shutdown()：不會立即終止執行緒池，而是要等所有任務快取佇列中的任務都執行完后才終止，但再也不會接受新的任務
shutdownNow()：立即終止執行緒池，并嘗試打斷正在執行的任務，并且清空任務快取佇列，回傳尚未執行的任務

3、執行緒池容量調整

ThreadPoolExecutor提供了動態調整執行緒池容量大小的方法：

setCorePoolSize：設定核心池大小
setMaximumPoolSize：設定執行緒池最大能創建的執行緒數目大小

當上述引數從小變大時，ThreadPoolExecutor進行執行緒賦值，還可能立即創建新的執行緒來執行任務，

使用執行緒池

ThreadPoolExecutor

通過構造方法使用ThreadPoolExecutor是執行緒池最直接的使用方式，下面看一個實體：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MyTest {
	public static void main(String[] args) {
		// 創建執行緒池
		ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3, 5, 5, TimeUnit.SECONDS,
				new ArrayBlockingQueue<Runnable>(5));
		// 向執行緒池提交任務
		for (int i = 0; i < threadPool.getCorePoolSize(); i++) {
			threadPool.execute(new Runnable() {
				@Override
				public void run() {
					for (int x = 0; x < 2; x++) {
						System.out.println(Thread.currentThread().getName() + ":" + x);
						try {
							Thread.sleep(2000);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
				}
			});
		}

		// 關閉執行緒池
		threadPool.shutdown(); // 設定執行緒池的狀態為SHUTDOWN，然后中斷所有沒有正在執行任務的執行緒
		// threadPool.shutdownNow(); // 設定執行緒池的狀態為STOP，然后嘗試停止所有的正在執行或暫停任務的執行緒，并回傳等待執行任務的串列，該方法要慎用，容易造成不可控的后果
	}
}

運行結果：

pool-1-thread-2:0
pool-1-thread-1:0
pool-1-thread-3:0
pool-1-thread-2:1
pool-1-thread-3:1
pool-1-thread-1:1

Executors封裝執行緒池

另外，Executors封裝好了4種常見的功能執行緒池（還是那么地貼心）：

1、FixedThreadPool

固定容量執行緒池，其特點是最大執行緒數就是核心執行緒數，意味著執行緒池只能創建核心執行緒，keepAliveTime為0，即執行緒執行完任務立即回收，任務佇列未指定容量，代表使用默認值Integer.MAX_VALUE，適用于需要控制并發執行緒的場景，

// 使用默認執行緒工廠
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
// 需要自定義執行緒工廠
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>(),
                                  threadFactory);
}

使用示例：

// 1. 創建執行緒池物件，設定核心執行緒和最大執行緒數為5
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
// 2. 創建Runnable（任務）
Runnable task =new Runnable(){
  public void run() {
     System.out.println(Thread.currentThread().getName() + "--->運行");
  }
};
// 3. 向執行緒池提交任務
fixedThreadPool.execute(task);

2、 SingleThreadExecutor

單執行緒執行緒池，特點是執行緒池中只有一個執行緒（核心執行緒），執行緒執行完任務立即回收，使用有界阻塞佇列（容量未指定，使用默認值Integer.MAX_VALUE）

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
// 為節省篇幅，省略了自定義執行緒工廠方式的原始碼

使用示例：

// 1. 創建單執行緒執行緒池
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
// 2. 創建Runnable（任務）
Runnable task = new Runnable(){
  public void run() {
     System.out.println(Thread.currentThread().getName() + "--->運行");
  }
};
// 3. 向執行緒池提交任務
singleThreadExecutor.execute(task);

3、 ScheduledThreadPool

定時執行緒池，指定核心執行緒數量，普通執行緒數量無限，執行緒執行完任務立即回收，任務佇列為延時阻塞佇列，這是一個比較特別的執行緒池，適用于執行定時或周期性的任務，

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

// 繼承了 ThreadPoolExecutor
public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor
        implements ScheduledExecutorService {
    // 建構式，省略了自定義執行緒工廠的建構式
	public ScheduledThreadPoolExecutor(int corePoolSize) {
    	super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          	new DelayedWorkQueue());
	}
	
	// 延時執行任務
	public ScheduledFuture<?> schedule(Runnable command,
                                       long delay,
                                       TimeUnit unit) {
        ...
    }
	// 定時執行任務
	public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit) {...}
}

使用示例：

// 1. 創建定時執行緒池
ExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
// 2. 創建Runnable（任務）
Runnable task = new Runnable(){
  public void run() {
     System.out.println(Thread.currentThread().getName() + "--->運行");
  }
};
// 3. 向執行緒池提交任務
scheduledThreadPool.schedule(task, 2, TimeUnit.SECONDS); // 延遲2s后執行任務
scheduledThreadPool.scheduleAtFixedRate(task,50,2000,TimeUnit.MILLISECONDS);// 延遲50ms后、每隔2000ms執行任務

4、CachedThreadPool

快取執行緒池，沒有核心執行緒，普通執行緒數量為Integer.MAX_VALUE（可以理解為無限），執行緒閑置60s后回收，任務佇列使用SynchronousQueue這種無容量的同步佇列，適用于任務量大但耗時低的場景，

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

使用示例：

// 1. 創建快取執行緒池
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
// 2. 創建Runnable（任務）
Runnable task = new Runnable(){
  public void run() {
     System.out.println(Thread.currentThread().getName() + "--->運行");
  }
};
// 3. 向執行緒池提交任務
cachedThreadPool.execute(task);

解讀執行緒池

OK，相信前面內容閱讀起來還算輕松愉悅吧，那么從這里開始就進入深水區了，如果后面內容能吃透，那么執行緒池知識就真的被你掌握了，

我們知道，向執行緒池提交任務是用ThreadPoolExecutor的execute()方法，但在其內部，執行緒任務的處理其實是相當復雜的，涉及到ThreadPoolExecutor、Worker、Thread三個類的6個方法：
在這里插入圖片描述

execute()

在ThreadPoolExecutor類中，任務提交方法的入口是execute(Runnable command)方法（submit()方法也是呼叫了execute()），該方法其實只在嘗試做一件事：經過各種校驗之后，呼叫 addWorker(Runnable command,boolean core)方法為執行緒池創建一個執行緒并執行任務，與之相對應，execute() 的結果有兩個：

引數說明：

Runnable command：待執行的任務

執行流程：

1、通過 ctl.get() 得到執行緒池的當前執行緒數，如果執行緒數小于corePoolSize，則呼叫 addWorker(commond,true)方法創建新的執行緒執行任務，否則執行步驟2；

2、步驟1失敗，說明已經無法再創建新執行緒，那么考慮將任務放入阻塞佇列，等待執行完任務的執行緒來處理，基于此，判斷執行緒池是否處于Running狀態（只有Running狀態的執行緒池可以接受新任務），如果任務添加到任務佇列成功則進入步驟3，失敗則進入步驟4；

3、來到這一步需要說明任務已經加入任務佇列，這時要二次校驗執行緒池的狀態，會有以下情形：

執行緒池不再是Running狀態了，需要將任務從任務佇列中移除，如果移除成功則拒絕本次任務
執行緒池是Running狀態，則判斷執行緒池作業執行緒是否為0，是則呼叫 addWorker(commond,true)添加一個沒有初始任務的執行緒（這個執行緒將去獲取已經加入任務佇列的本次任務并執行），否則進入步驟4；
執行緒池不是Running狀態，但從任務佇列移除任務失敗（可能已被某執行緒獲取？），進入步驟4；

4、將執行緒池擴容至maximumPoolSize并呼叫 addWorker(commond,false)方法創建新的執行緒執行任務，失敗則拒絕本次任務，

流程圖：
在這里插入圖片描述
原始碼詳讀：

/**
 * 在將來的某個時候執行給定的任務，任務可以在新執行緒中執行，也可以在現有的池執行緒中執行，
 * 如果由于此執行器已關倍訓已達到其容量而無法提交任務以供執行，則由當前的{@code RejectedExecutionHandler}處理該任務，
 * 
 * @param command the task to execute  待執行的任務命令
 */
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     * 
     * 1. 如果運行的執行緒少于corePoolSize，將嘗試以給定的命令作為第一個任務啟動新執行緒，
     *
     * 2. 如果一個任務可以成功排隊，那么我們仍然需要仔細檢查兩點，其一，我們是否應該添加一個執行緒
     * （因為自從上次檢查至今，一些存在的執行緒已經死亡），其二，執行緒池狀態此時已改變成非運行態，因此，我們重新檢查狀態，如果檢查不通過，則移除已經入列的任務，如果檢查通過且執行緒池執行緒數為0，則啟動新執行緒，
     * 
     * 3. 如果無法將任務加入任務佇列，則將執行緒池擴容到極限容量并嘗試創建一個新執行緒，如果失敗則拒絕任務，
     */
    int c = ctl.get();
   
    // 步驟1：判斷執行緒池當前執行緒數是否小于執行緒池大小
    if (workerCountOf(c) < corePoolSize) {
        // 增加一個作業執行緒并添加任務，成功則回傳，否則進行步驟2
        // true代表使用coreSize作為邊界約束，否則使用maximumPoolSize
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 步驟2：不滿足workerCountOf(c) < corePoolSize或addWorker失敗，進入步驟2
    // 校驗執行緒池是否是Running狀態且任務是否成功放入workQueue（阻塞佇列）
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 再次校驗，如果執行緒池非Running且從任務佇列中移除任務成功，則拒絕該任務
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 如果執行緒池作業執行緒數量為0，則新建一個空任務的執行緒
        else if (workerCountOf(recheck) == 0)
            // 如果執行緒池不是Running狀態，是加入不進去的
            addWorker(null, false);
    }
    // 步驟3：如果執行緒池不是Running狀態或任務入列失敗，嘗試擴容maxPoolSize后再次addWorker，失敗則拒絕任務
    else if (!addWorker(command, false))
        reject(command);
}

addWorker()

addWorker(Runnable firstTask, boolean core) 方法，顧名思義，向執行緒池添加一個帶有任務的作業執行緒，

引數說明：

Runnable firstTask：新創建的執行緒應該首先運行的任務（如果沒有，則為空），
boolean core：該引數決定了執行緒池容量的約束條件，即當前執行緒數量以何值為極限值，引數為 true 則使用corePollSize 作為約束值，否則使用maximumPoolSize，

執行流程：

1、外層回圈判斷執行緒池的狀態是否可以新增作業執行緒，這層校驗基于下面兩個原則：

執行緒池為Running狀態時，既可以接受新任務也可以處理任務
執行緒池為關閉狀態時只能新增空任務的作業執行緒（worker）處理任務佇列（workQueue）中的任務不能接受新任務

2、內層回圈向執行緒池添加作業執行緒并回傳是否添加成功的結果，

首先校驗執行緒數是否已經超限制，是則回傳false，否則進入下一步
通過CAS使作業執行緒數+1，成功則進入步驟3，失敗則再次校驗執行緒池是否是運行狀態，是則繼續內層回圈，不是則回傳外層回圈

3、核心執行緒數量+1成功的后續操作：添加到作業執行緒集合，并啟動作業執行緒

首先獲取鎖之后，再次校驗執行緒池狀態（具體校驗規則見代碼注解），通過則進入下一步，未通過則添加執行緒失敗
執行緒池狀態校驗通過后，再檢查執行緒是否已經啟動，是則拋出例外，否則嘗試將執行緒加入執行緒池
檢查執行緒是否啟動成功，成功則回傳true，失敗則進入 addWorkerFailed 方法

流程圖：
在這里插入圖片描述
原始碼詳讀：

private boolean addWorker(Runnable firstTask, boolean core) {
    // 外層回圈：判斷執行緒池狀態
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        /** 
         * 1.執行緒池為非Running狀態（Running狀態則既可以新增核心執行緒也可以接受任務）
         * 2.執行緒為shutdown狀態且firstTask為空且佇列不為空
         * 3.滿足條件1且條件2不滿足，則回傳false
         * 4.條件2解讀：執行緒池為shutdown狀態時且任務佇列不為空時，可以新增空任務的執行緒來處理佇列中的任務
         */
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

		// 內層回圈：執行緒池添加核心執行緒并回傳是否添加成功的結果
        for (;;) {
            int wc = workerCountOf(c);
            // 校驗執行緒池已有執行緒數量是否超限：
            // 1.執行緒池最大上限CAPACITY 
            // 2.corePoolSize或maximumPoolSize（取決于入參core）
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize)) 
                return false;
            // 通過CAS操作使作業執行緒數+1，跳出外層回圈
            if (compareAndIncrementWorkerCount(c)) 
                break retry;
            // 執行緒+1失敗，重讀ctl
            c = ctl.get();   // Re-read ctl
            // 如果此時執行緒池狀態不再是running，則重新進行外層回圈
            if (runStateOf(c) != rs)
                continue retry;
            // 其他 CAS 失敗是因為作業執行緒數量改變了，繼續內層回圈嘗試CAS對執行緒數+1
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    /**
     * 核心執行緒數量+1成功的后續操作：添加到作業執行緒集合，并啟動作業執行緒
     */
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        final ReentrantLock mainLock = this.mainLock;
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            // 下面代碼需要加鎖：執行緒池主鎖
            mainLock.lock(); 
            try {
                // 持鎖期間重新檢查，執行緒工廠創建執行緒失敗或獲取鎖之前關閉的情況發生時，退出
                int c = ctl.get();
                int rs = runStateOf(c);

				// 再次檢驗執行緒池是否是running狀態或執行緒池shutdown但執行緒任務為空
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    // 執行緒已經啟動，則拋出非法執行緒狀態例外
                    // 為什么會存在這種狀態呢？未解決
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    workers.add(w); //加入執行緒池
                    int s = workers.size();
                    // 如果當前作業執行緒數超過執行緒池曾經出現過的最大執行緒數，重繪后者值
                    if (s > largestPoolSize)
                        largestPoolSize = s; 
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();  // 釋放鎖
            }
            if (workerAdded) { // 作業執行緒添加成功，啟動該執行緒
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        //執行緒啟動失敗，則進入addWorkerFailed
        if (! workerStarted) 
            addWorkerFailed(w);
    }
    return workerStarted;
}

Worker類

Worker類是內部類，既實作了Runnable，又繼承了AbstractQueuedSynchronizer（以下簡稱AQS），所以其既是一個可執行的任務，又可以達到鎖的效果，

Worker類主要維護正在運行任務的執行緒的中斷控制狀態，以及其他次要的記錄，這個類適時地繼承了AbstractQueuedSynchronizer類，以簡化獲取和釋放鎖（該鎖作用于每個任務執行代碼）的程序，這樣可以防止去中斷正在運行中的任務，只會中斷在等待從任務佇列中獲取任務的執行緒，

我們實作了一個簡單的不可重入互斥鎖，而不是使用可重入鎖，因為我們不希望作業任務在呼叫setCorePoolSize之類的池控制方法時能夠重新獲取鎖，另外，為了在執行緒真正開始運行任務之前禁止中斷，我們將鎖狀態初始化為負值，并在啟動時清除它（在runWorker中），

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;
 
    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread; 
     
    /** Initial task to run.  Possibly null. */
    Runnable firstTask;
     
    /** Per-thread task counter */
    volatile long completedTasks;
 
    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
    // 通過建構式初始化，
    Worker(Runnable firstTask) {
        //設定AQS的同步狀態
        // state：鎖狀態，-1為初始值，0為unlock狀態，1為lock狀態
        setState(-1); // inhibit interrupts until runWorker  在呼叫runWorker前，禁止中斷
       
        this.firstTask = firstTask;
        // 執行緒工廠創建一個執行緒
        this.thread = getThreadFactory().newThread(this); 
    }
 
    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this); //runWorker()是ThreadPoolExecutor的方法
    }
 
    // Lock methods
    // The value 0 represents the unlocked state. 0代表“沒被鎖定”狀態
    // The value 1 represents the locked state. 1代表“鎖定”狀態
 
    protected boolean isHeldExclusively() {
        return getState() != 0;
    }
 
    /**
     * 嘗試獲取鎖的方法
     * 重寫AQS的tryAcquire()，AQS本來就是讓子類來實作的
     */
    protected boolean tryAcquire(int unused) {
        // 判斷原值為0，且重置為1，所以state為-1時，鎖無法獲取，
        // 每次都是0->1，保證了鎖的不可重入性
        if (compareAndSetState(0, 1)) {
            // 設定exclusiveOwnerThread=當前執行緒
            setExclusiveOwnerThread(Thread.currentThread()); 
            return true;
        }
        return false;
    }
 
    /**
     * 嘗試釋放鎖
     * 不是state-1，而是置為0
     */
    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null); 
        setState(0);
        return true;
    }
 
    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }
 
    /**
     * 中斷（如果運行）
     * shutdownNow時會回圈對worker執行緒執行
     * 且不需要獲取worker鎖，即使在worker運行時也可以中斷
     */
    void interruptIfStarted() {
        Thread t;
        //如果state>=0、t!=null、且t沒有被中斷
        //new Worker()時state==-1，說明不能中斷
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

runWorker()

可以說，runWorker(Worker w) 是執行緒池中真正處理任務的方法，前面的execute() 和 addWorker() 都是在為該方法做準備和鋪墊，

引數說明：

Worker w：封裝的Worker，攜帶了作業執行緒的諸多要素，包括Runnable（待處理任務）、lock（鎖）、completedTasks（記錄執行緒池已完成任務數）

執行流程：

1、判斷當前任務或者從任務佇列中獲取的任務是否不為空，都為空則進入步驟2，否則進入步驟3

2、任務為空，則將completedAbruptly置為false（即執行緒不是突然終止），并執行processWorkerExit(w,completedAbruptly)方法進入執行緒退出程式

3、任務不為空，則進入回圈，并加鎖

4、判斷是否為執行緒添加中斷標識，以下兩個條件滿足其一則添加中斷標識：

執行緒池狀態>=STOP,即STOP或TERMINATED
一開始判斷執行緒池狀態<STOP，接下來檢查發現Thread.interrupted()為true，即執行緒已經被中斷，再次檢查執行緒池狀態是否>=STOP（以消除該瞬間shutdown方法生效，使執行緒池處于STOP或TERMINATED）

5、執行前置方法 beforeExecute(wt, task)（該方法為空方法，由子類實作）后執行task.run() 方法執行任務（執行不成功拋出相應例外）

6、執行后置方法 afterExecute(task, thrown)（該方法為空方法，由子類實作）后將執行緒池已完成的任務數+1，并釋放鎖，

7、再次進行回圈條件判斷，

流程圖：
在這里插入圖片描述
原始碼詳讀：

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    // allow interrupts
    // new Worker()是state==-1，此處是呼叫Worker類的tryRelease()方法，將state置為0，而interruptIfStarted()中只有state>=0才允許呼叫中斷
    w.unlock(); 
            
    // 執行緒退出的原因，true是任務導致，false是執行緒正常退出
    boolean completedAbruptly = true; 
    try {
        // 當前任務和從任務佇列中獲取的任務都為空，方停止回圈
        while (task != null || (task = getTask()) != null) {
            //上鎖可以防止在shutdown()時終止正在運行的worker，而不是應對并發
            w.lock(); 
             
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            /**
             * 判斷1：確保只有在執行緒處于stop狀態且wt未中斷時，wt才會被設定中斷標識
             * 條件1：執行緒池狀態>=STOP,即STOP或TERMINATED
             * 條件2：一開始判斷執行緒池狀態<STOP，接下來檢查發現Thread.interrupted()為true，即執行緒已經被中斷，再次檢查執行緒池狀態是否>=STOP（以消除該瞬間shutdown方法生效，使執行緒池處于STOP或TERMINATED），
             * 條件1與條件2任意滿意一個，且wt不是中斷狀態，則中斷wt，否則進入下一步
             */
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt(); //當前執行緒呼叫interrupt()中斷
             
            try {
                //執行前（空方法，由子類重寫實作）
                beforeExecute(wt, task);
                 
                Throwable thrown = null;
                try {
                    task.run();
                } 
                catch (RuntimeException x) {
                    thrown = x; throw x;
                } 
                catch (Error x) {
                    thrown = x; throw x;
                } 
                catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } 
                finally {
                    //執行后（空方法，由子類重寫實作）
                    afterExecute(task, thrown); 
                }
            } 
            finally {
                task = null; 
                w.completedTasks++; //完成任務數+1
                w.unlock(); //釋放鎖
            }
        }
        // 
        completedAbruptly = false;
    } 
    finally {
        //處理worker的退出
        processWorkerExit(w, completedAbruptly);
    }
}

5、getTask()

由函式呼叫關系圖可知，在ThreadPoolExecutor類的實作中，Runnable getTask() 方法是為void runWorker(Worker w)方法服務的，它的作用就是在任務佇列（workQueue）中獲取 task（Runnable），

引數說明：無引數

執行流程：

將timedOut（上次獲取任務是否超時）置為false（首次執行方法，無上次，自然為false），進入一個無限回圈
如果執行緒池為Shutdown狀態且任務佇列為空（執行緒池shutdown狀態可以處理任務佇列中的任務，不再接受新任務，這個是重點）或者執行緒池為STOP或TERMINATED狀態，則意味著執行緒池不必再獲取任務了，當前作業執行緒數量-1并回傳null，否則進入步驟3
如果執行緒池數量超限制或者時間超限且（任務佇列為慷訓當前執行緒數>1），則進入步驟4，否則進入步驟5，
移除作業執行緒，成功則回傳null，不成功則進入下輪回圈，
嘗試用poll() 或者 take()（具體用哪個取決于timed的值）獲取任務，如果任務不為空，則回傳該任務，如果為空，則將timeOut 置為 true進入下一輪回圈，如果獲取任務程序發生例外，則將 timeOut置為 false 后進入下一輪回圈，

流程圖：
在這里插入圖片描述
原始碼詳讀：

private Runnable getTask() {
    // 最新一次poll是否超時
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        /**
         * 條件1：執行緒池狀態SHUTDOWN、STOP、TERMINATED狀態
         * 條件2：執行緒池STOP、TERMINATED狀態或workQueue為空
         * 條件1與條件2同時為true，則workerCount-1，并且回傳null
         * 注：條件2是考慮到SHUTDOWN狀態的執行緒池不會接受任務，但仍會處理任務
         */
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        /**
         * 下列兩個條件滿足任意一個，則給當前正在嘗試獲取任務的作業執行緒設定阻塞時間限制（超時會被銷毀？不太確定這點），否則執行緒可以一直保持活躍狀態
         * 1.allowCoreThreadTimeOut：當前執行緒是否以keepAliveTime為超時時限等待任務
         * 2.當前執行緒數量已經超越了核心執行緒數
         */
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            
        // 兩個條件全部為true，則通過CAS使作業執行緒數-1，即剔除作業執行緒
        // 條件1：作業執行緒數大于maximumPoolSize，或（作業執行緒阻塞時間受限且上次在任務佇列拉取任務超時）
        // 條件2：wc > 1或任務佇列為空
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            // 移除作業執行緒，成功則回傳null，不成功則進入下輪回圈
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

	    // 執行到這里，說明已經經過前面重重校驗，開始真正獲取task了
        try {
            // 如果作業執行緒阻塞時間受限，則使用poll(),否則使用take()
            // poll()設定阻塞時間，而take()無時間限制，直到拿到結果為止
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            // r不為空，則回傳該Runnable
            if (r != null)
                return r;
            // 沒能獲取到Runable，則將最近獲取任務是否超時設定為true
            timedOut = true;
        } catch (InterruptedException retry) {
            // 回應中斷，進入下一次回圈前將最近獲取任務超時狀態置為false
            timedOut = false;
        }
    }
}

processWorkerExit()

processWorkerExit(Worker w, boolean completedAbruptly)執行執行緒退出的方法

引數說明：

Worker w：要結束的作業執行緒，
boolean completedAbruptly：是否突然完成（例外導致），如果作業執行緒因為用戶例外死亡，則completedAbruptly引數為 true，

執行流程：

1、如果 completedAbruptly 為 true，即作業執行緒因為例外突然死亡，則執行作業執行緒-1操作，

2、主執行緒獲取鎖后，執行緒池已經完成的任務數追加 w（當前作業執行緒）完成的任務數，并從worker的set集合中移除當前worker，

3、根據執行緒池狀態進行判斷是否執行tryTerminate()結束執行緒池，

4、是否需要增加作業執行緒，如果執行緒池還沒有完全終止，仍需要保持一定數量的執行緒，

如果當前執行緒是突然終止的，呼叫addWorker()創建作業執行緒
當前執行緒不是突然終止，但當前作業執行緒數量小于執行緒池需要維護的執行緒數量，則創建作業執行緒，需要維護的執行緒數量為corePoolSize（取決于成員變數 allowCoreThreadTimeOut是否為 false）或1，
原始碼詳讀：

/**
 * Performs cleanup and bookkeeping for a dying worker. Called
 * only from worker threads. Unless completedAbruptly is set,
 * assumes that workerCount has already been adjusted to account
 * for exit.  This method removes thread from worker set, and
 * possibly terminates the pool or replaces the worker if either
 * it exited due to user task exception or if fewer than
 * corePoolSize workers are running or queue is non-empty but
 * there are no workers.
 *
 * @param w the worker
 * @param completedAbruptly if the worker died due to user exception
 */
private void processWorkerExit(Worker w, boolean completedAbruptly) {
    /**
     * 1.作業執行緒-1操作
     * 1）如果completedAbruptly 為true，說明作業執行緒發生例外，那么將正在作業的執行緒數量-1
     * 2）如果completedAbruptly 為false，說明作業執行緒無任務可以執行，由getTask()執行worker-1操作
     */
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();

    // 2.從執行緒set集合中移除作業執行緒，該程序需要加鎖
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 將該worker已完成的任務數追加到執行緒池已完成的任務數
        completedTaskCount += w.completedTasks;
        // HashSet<Worker>中移除該worker
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    
	// 3.根據執行緒池狀態進行判斷是否結束執行緒池
    tryTerminate();
	
	/**
     * 4.是否需要增加作業執行緒
     * 執行緒池狀態是running 或 shutdown
     * 如果當前執行緒是突然終止的，addWorker()
     * 如果當前執行緒不是突然終止的，但當前執行緒數量 < 要維護的執行緒數量，addWorker()
     * 故如果呼叫執行緒池shutdown()，直到workQueue為空前，執行緒池都會維持corePoolSize個執行緒，然后再逐漸銷毀這corePoolSize個執行緒
     */
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
       if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/265450.html

標籤：其他

上一篇：Linux執行緒的創建與回收

下一篇：層次分析法（AHP）原理以及應用

深入Java執行緒池：從設計思想到原始碼解讀

執行緒池：從設計思想到原始碼決議

前言

初識執行緒池

執行緒池優勢

執行緒池設計思路

深入執行緒池

構造方法

任務佇列

拒絕策略

執行緒池狀態

初始化&容量調整&關閉

使用執行緒池

ThreadPoolExecutor

Executors封裝執行緒池

解讀執行緒池

execute()

addWorker()

Worker類

runWorker()

processWorkerExit()