2. 程式人生 > 實用技巧 >程池中狀態與執行緒數的設計分析（ThreadPoolExecutor中ctl的設計分析）

程池中狀態與執行緒數的設計分析（ThreadPoolExecutor中ctl的設計分析）

阿新 • • 發佈：2020-07-04

目錄

預備知識

可以先看下我的另一篇文章對於Java中的位掩碼BitMask的解釋。

  • 1、一個整數在jvm中佔用了4個位元組，共32bits
  • 2、最高位的bit代表符號位，0為正數、1為負，剩餘的31bits則代表數字部分
  • 3、反碼加1即為補碼
  • 4、對於負數而言，是以補碼的形式儲存在記憶體中的。以-7（int）為例
    • 1)、將-7的絕對值轉化為二進位制：
      0000 0000 0000 0000 0000 0000 0000 0111
    • 2)：將上面的二進位制以反碼錶示：
      1111 1111 1111 1111 1111 1111 1111 1000
    • 3)：轉化為補碼：
      1111 1111 1111 1111 1111 1111 1111 1001

原始碼分析

我們把ThreadPoolExecutor中的狀態和狀態相關的方法複製出來，然後建立一個執行緒池，在執行中的時候分析執行緒池的狀態和執行緒數，於是有了下面例子：

@Slf4j
public class ThreadPoolExecutorCtlAnalysis {
    private static final int COUNT_BITS = Integer.SIZE - 3;
    private static final int CAPACITY = (1 << COUNT_BITS) - 1;// 000,11111111111111111111111111111

    // runState is stored in the high-order bits
    private static final int RUNNING = -1 << COUNT_BITS;  // 111,00000000000000000000000000000
    private static final int SHUTDOWN = 0 << COUNT_BITS;  // 000,00000000000000000000000000000
    private static final int STOP = 1 << COUNT_BITS;      // 001,00000000000000000000000000000
    private static final int TIDYING = 2 << COUNT_BITS;   // 010,00000000000000000000000000000
    private static final int TERMINATED = 3 << COUNT_BITS;// 011,00000000000000000000000000000

    // Packing and unpacking ctl

    // RUNNING(3'thread) 111,00000000000000000000000000011
    // ~CAPACITY         111,00000000000000000000000000000
    // RESULT            111,00000000000000000000000000000
    // 與操作取高位獲取的就是ctl中儲存的的執行緒池的狀態
    private static int runStateOf(int c) {
        return c & ~CAPACITY;
    }

    // RUNNING(3'thread) 111,00000000000000000000000000011
    // CAPACITY          000,11111111111111111111111111111
    // RESULT            000,00000000000000000000000000011
    // 與操作取低位獲取的就是ctl中儲存的worker數量
    private static int workerCountOf(int c) {
        return c & CAPACITY;
    }

    private static Runnable buildRunnableTask() {
        return () -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info("Task finished.");
        };
    }

    private static int getCtlValue(ThreadPoolExecutor executor, Field field) {
        //noinspection ConstantConditions
        return ((AtomicInteger) ReflectionUtils.getField(field, executor)).get();
    }

    private static String formatBinaryString(int state) {
        StringBuilder binaryString = new StringBuilder(Integer.toBinaryString(state));
        if (binaryString.length() < Integer.SIZE) {
            for (int i = binaryString.length(); i < Integer.SIZE; i++) {
                binaryString.insert(0, "0");
            }
        }
        return binaryString.substring(0, 3) + "," + binaryString.substring(3, Integer.SIZE);
    }

    private static void peekThreadPoolExecuteState(ThreadPoolExecutor executor, Field ctlField) {
        log.info("------------------- ThreadPoolExecuteState -------------------");
        int ctlValue = getCtlValue(executor, ctlField);
        log.info("getCtlValue  : {}", formatBinaryString(ctlValue));
        log.info("workerCountOf: {}", workerCountOf(ctlValue));
        log.info("Is    RUNNING: {}", runStateOf(ctlValue) == RUNNING);
        log.info("Is   SHUTDOWN: {}", runStateOf(ctlValue) == SHUTDOWN);
        log.info("Is       STOP: {}", runStateOf(ctlValue) == STOP);
        log.info("Is    TIDYING: {}", runStateOf(ctlValue) == TIDYING);
        log.info("Is TERMINATED: {}", runStateOf(ctlValue) == TERMINATED);
    }

    public static void main(String[] args) throws NoSuchFieldException, InterruptedException {
        // 打印出來看看幾種狀態的二進位制表示
        log.info("{} --> CAPACITY", formatBinaryString(CAPACITY));
        log.info("{} --> RUNNING", formatBinaryString(RUNNING));
        log.info("{} --> STOP", formatBinaryString(STOP));
        log.info("{} --> TERMINATED", formatBinaryString(TERMINATED));


        // 建立一個執行緒池，執行兩個任務
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1, 2, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(1));
        executor.submit(buildRunnableTask());
        executor.submit(buildRunnableTask());
        executor.submit(buildRunnableTask());
        // 休眠一秒鐘，可以拿到中間狀態的ctl
        Thread.sleep(1000);
        log.info("getActiveCount(): {}", executor.getActiveCount());
        // 通過反射能拿到ThreadPoolExecutor的ctl的值
        Field ctlField = ThreadPoolExecutor.class.getDeclaredField("ctl");
        ctlField.setAccessible(true);
        // 執行緒池執行中的狀態可通過ctl拿到
        peekThreadPoolExecuteState(executor, ctlField);
        // 終止執行緒池，再來看看執行緒池中ctl的狀態
        executor.shutdownNow();
        peekThreadPoolExecuteState(executor, ctlField);
        // 休眠2秒鐘，看看執行緒池最終的狀態
        Thread.sleep(2000);
        peekThreadPoolExecuteState(executor, ctlField);
    }
}

在看執行結果之前，我們先看下ThreadPoolExecutor中的幾處涉及到狀態變更的方法實現。

submit()原始碼分析

public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}

最終呼叫的是內部的execute方法：

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);
}

這個方法不是特別複雜，我們本文的重點是要看看它的addWorker()方法，這個不復制太多邏輯，關鍵在兩行：

private boolean addWorker(Runnable firstTask, boolean core) {
    int c = ctl.get();
    ...
        compareAndIncrementWorkerCount(c)
    ...
}
private boolean compareAndIncrementWorkerCount(int expect) {
    return ctl.compareAndSet(expect, expect + 1);
}

這裡控制的是ctl中工作執行緒數(wc:WorkerCount)的變更，即整形低29位的自增不會影響到高3位的狀態：

RUNNING(0'wc) 111,00000000000000000000000000000
RUNNING(1'wc) 111,00000000000000000000000000001

所以可預見的輸出結果就是：

workerCountOf(): 1
Is Running: true
Is Stop: false

注意的是這些值都從ctl屬性中得來。

shutdownNow()原始碼分析

在我們的例子中，我們呼叫了shutdownNow()方法來改變執行緒池的狀態。

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(STOP);
        interruptWorkers();
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}

這裡我們關注的是advanceRunState(STOP)方法：

/**
 * Transitions runState to given target, or leaves it alone if
 * already at least the given target.
 *
 * @param targetState the desired state, either SHUTDOWN or STOP
 *        (but not TIDYING or TERMINATED -- use tryTerminate for that)
 */
private void advanceRunState(int targetState) {
    for (;;) {
        int c = ctl.get();
        if (runStateAtLeast(c, targetState) ||
            ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
            break;
    }
}

該方法最終是要把當前狀態變為STOP狀態。

注意點一：

ThreadPoolExecutor中狀態定義的值大小是有序的，即：

TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING（最高位1是負數）

注意點二：

advance的含意是推進、前進的意思，Java併發包裡的很多方法都使用了該命名，所以當前方法表示的意思是要推進執行狀態（advanceRunState），因此方法中才有了runStateAtLeast()判斷。
即要推進狀態，那如果當前狀態已經大於目標狀態了，本次方法直接跳過。否則才去做cas操作。這也正是原方法註釋想表達的意思。

注意點三：

最後在做cas操作的時候合併當前wc和rs的值，使用的是ctlOf方法：

private static int ctlOf(int rs, int wc) {
    return rs | wc;
}

通過與運算把RunState和WorkerCount的值合併到一處，即最終的ctl的值：

STOP      001,00000000000000000000000000000
WorkCount 000,00000000000000000000000000001
ctl value 001,00000000000000000000000000001

所以可預見的輸出結果就是：

workerCountOf(): 1
Is Running: false
Is Stop: true

程式碼輸出

12:56:19.473 [main] ThreadPoolExecutorCtlAnalysis - 000,11111111111111111111111111111 --> CAPACITY
12:56:19.476 [main] ThreadPoolExecutorCtlAnalysis - 111,00000000000000000000000000000 --> RUNNING
12:56:19.476 [main] ThreadPoolExecutorCtlAnalysis - 001,00000000000000000000000000000 --> STOP
12:56:19.476 [main] ThreadPoolExecutorCtlAnalysis - 011,00000000000000000000000000000 --> TERMINATED
12:56:20.520 [main] ThreadPoolExecutorCtlAnalysis - getActiveCount(): 2
12:56:20.520 [main] ThreadPoolExecutorCtlAnalysis - ------------------- ThreadPoolExecuteState -------------------
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - getCtlValue  : 111,00000000000000000000000000010
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - workerCountOf: 2
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - Is    RUNNING: true
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - Is   SHUTDOWN: false
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - Is       STOP: false
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - Is    TIDYING: false
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - Is TERMINATED: false
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - ------------------- ThreadPoolExecuteState -------------------
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - getCtlValue  : 001,00000000000000000000000000010
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - workerCountOf: 2
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - Is    RUNNING: false
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - Is   SHUTDOWN: false
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - Is       STOP: true
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - Is    TIDYING: false
12:56:20.533 [main] ThreadPoolExecutorCtlAnalysis - Is TERMINATED: false
12:56:20.534 [pool-1-thread-1] ThreadPoolExecutorCtlAnalysis - Task finished.
12:56:20.534 [pool-1-thread-2] ThreadPoolExecutorCtlAnalysis - Task finished.
12:56:22.538 [main] ThreadPoolExecutorCtlAnalysis - ------------------- ThreadPoolExecuteState -------------------
12:56:22.538 [main] ThreadPoolExecutorCtlAnalysis - getCtlValue  : 011,00000000000000000000000000000
12:56:22.538 [main] ThreadPoolExecutorCtlAnalysis - workerCountOf: 0
12:56:22.539 [main] ThreadPoolExecutorCtlAnalysis - Is    RUNNING: false
12:56:22.539 [main] ThreadPoolExecutorCtlAnalysis - Is   SHUTDOWN: false
12:56:22.539 [main] ThreadPoolExecutorCtlAnalysis - Is       STOP: false
12:56:22.539 [main] ThreadPoolExecutorCtlAnalysis - Is    TIDYING: false
12:56:22.539 [main] ThreadPoolExecutorCtlAnalysis - Is TERMINATED: true

可以看到使用ctl一個欄位可以獲取到兩個值，並且這兩個值不會有併發不一致的情況，每次都是一次cas更新值。

設計目的與優點

執行緒池自身的狀態和執行緒數量都維護在一個原子變數ctl中，目的不是為了減少儲存空間，而是將執行緒池狀態與執行緒個數合二為一，這樣就可以用一次cas原子操作進行賦值，更容易保證在多執行緒環境下保證執行狀態和執行緒數量的統一。這真是大師的設計智慧啊！

相關推薦

程池中狀態與執行緒數的設計分析（ThreadPoolExecutor中ctl的設計分析）

目錄預備知識原始碼分析submit()原始碼分析shutdownNow()原始碼分析程式碼輸出設計目的與優點

執行緒狀態與執行緒池狀態

執行緒狀態 NEW、RUNNABLE、BLOCKED、WAITING、TIMED_WAITING、TERMINATED 參考java.lang.Thread類的內部列舉java.lang.Thread.State，如下：

【轉載】嘗試解決在建構函式中同步呼叫Dns.GetHostAddressesAsync()引起的執行緒死鎖 .NET Core中遇到奇怪的執行緒死鎖問題：記憶體與執行緒數不停地增長

（最終採用的是方法4） 問題詳情見：.NET Core中遇到奇怪的執行緒死鎖問題：記憶體與執行緒數不停地增長

JAVA 自定義執行緒池的最大執行緒數設定方法

一：CPU密集型： 　　定義：CPU密集型也是指計算密集型，大部分時間用來做計算邏輯判斷等CPU動作的程式稱為CPU密集型任務。該型別的任務需要進行大量的計算，主要消耗CPU資源。 這種計算密集型任務雖然也可以用多工完

執行緒五大狀態與執行緒停止

執行緒五大狀態與執行緒停止 執行緒狀態 執行緒方法 停止執行緒 不推薦使用JDK提供的stop()、destroy()方法。【已廢棄】

Java多執行緒-執行緒池的使用與執行緒總結（狂神說含程式碼）

使用執行緒池 背景：經常建立和銷燬、使用量特別大的資源，比如併發情況下的執行緒，對效能影響很大

簡述Linux 中程序與執行緒

程序是一個具有一定獨立功能的程式關於某個資料集合的一次執行活動。它是作業系統動態執行的基本單元，在傳統的作業系統中，程序既是基本的分配單元，也是基本的執行單元。

【程序與執行緒】Linux中程序與執行緒的區別

1.執行緒的建立方法 建立執行緒具體呼叫pthread_create函式,這個函式實在glibc庫中實現。在glibc中pthread_create的呼叫路徑是__pthread_create_2_1->create_thread。其中create_thread很重要，它設定了建立執行緒

執行緒與執行緒鎖---python版本（附帶執行緒鎖例項）

程序和執行緒 現代的作業系統基本上都是支援多工的作業系統 那？什麼叫多工呢，簡單而言，就是作業系統可以同時執行多個任務。

同事臨走時，給了我這份多執行緒and高併發（面試題+思維導圖），藉此我含淚拿下了阿里offer

導語： 作為一名Java開發者，聽到多執行緒應該不會感到陌生吧，多執行緒也是在面試時面試官必問的一個問題了，不懂多執行緒還敢說自己是做Java開發的嗎？

Qt實現基於多執行緒的檔案傳輸（服務端，客戶端）

1. 效果 先看看效果圖 這是傳輸檔案完成的介面 客戶端 服務端 2. 知識準備 其實檔案傳輸和聊天室十分相似，只不過一個傳輸的是文字，一個傳輸的是檔案，而這方面的知識，我已經在前面的部落格寫過了，不瞭解的同

深入理解-CPU核心數與執行緒池併發執行緒數關係

那是一個風和日麗的下午！ 面試官微微一笑，對我說：“小夥子，合理配置執行緒池你是如何考慮的？”

python Event事件、程序池與執行緒池、協程解析

Event事件 用來控制執行緒的執行 出現e.wait()，就會把這個執行緒設定為False，就不能執行這個任務；

SpringBoot中的非同步操作與執行緒池

SpringBoot中的非同步操作與執行緒池 執行緒池型別 Java通過 java.util.concurrent.Executors 的靜態方法提供五種執行緒池

看完這篇還不懂高併發中的執行緒與執行緒池你來打我

從這篇開始將會開啟高效能、高併發系列，本篇是該系列的開篇，主要關注多執行緒以及執行緒池。

執行緒池大小設定，CPU的核心數、執行緒數的關係和區別，同步與堵塞完全是兩碼事

執行緒池大小設定，CPU的核心數、執行緒數的關係和區別，同步與堵塞完全是兩碼事

Qt中的多執行緒與執行緒池淺析+例項

1. Qt中的多執行緒與執行緒池 今天學習了Qt中的多執行緒和執行緒池，特寫這篇部落格來記錄一下

GIL與普通互斥鎖區別，死鎖現象，訊號量，event事件，程序池與執行緒池，協程

GIL與普通互斥鎖區別 GIL鎖和互斥鎖的異同點 相同： 都是為了解決直譯器中多個執行緒資源競爭的問題

程序池與執行緒池 協程相關知識

概要 GIL與普通互斥鎖區別 驗證多執行緒是否有用 死鎖現象(強調鎖不能輕易使用)

GIL與普通互斥鎖區別 驗證多執行緒 死鎖現象 訊號量和event事件 程序池與執行緒池 協程

day37   GIL與普通互斥鎖區別 GIL： 　　在單核CPU上執行多個執行緒時，使用GIL可以保證系統資源只分配給一個執行緒，其它執行緒在GIL釋放之前保持等待狀態。