预备知识
可以先看下我的另一篇文章对于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原子操作进行赋值，更容易保证在多线程环境下保证运行状态和线程数量的统一。这真是大师的设计智慧啊！
网友评论