預備知識
可以先看下我的另一篇文章對於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
- 1)、將-7的絕對值轉化為二進制：
源碼分析
我們把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));
// 創建一個線程池，運行3個任務
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原子操作進行賦值，更容易保證在多線程環境下保證運行狀態和線程數量的統一。這真是大師的設計智慧啊！