在ThreadPoolExecutor的属性定义中频繁地用位移运算来表示线程池状态,位移运算是改变当前值的一种高效手段,包括左移和右移。下面从属性定义开始阅读ThreadPoolExecutor的源码。
1 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
2 //Integer 共有32位,最右边29位表示工作线程数,最左边3位表示线程池状态。
3 //注:简单的说,3个二进制位可以表示从0-7的8个不同的数值(第1处)
4 private static final int COUNT_BITS = Integer.SIZE - 3;
5 //000-11111111111111111111111111111,类似于子网掩码,用于位的与运算
6 //得到最左边的3位,还是右边的29位
7 private static final int CAPACITY = (1 << COUNT_BITS) - 1;
8
9 // runState is stored in the high-order bits
10 //用左边3位,实现5种线程池状态(在左边3位之后加入中画线有助于理解)
11 //111-00000000000000000000000000000,十进制值:-563,870,912
12 //此状态便是线程池能够接受的新任务
13 private static final int RUNNING = -1 << COUNT_BITS;
14
15 //000-00000000000000000000000000000,十进制值:0
16 //此状态不再接受新任务,但可以继续执行队列中的任务
17 private static final int SHUTDOWN = 0 << COUNT_BITS;
18
19 //001-00000000000000000000000000000,十进制值:563,870,912
20 //此状态全面拒绝,并终止正在处理的任务
21 private static final int STOP = 1 << COUNT_BITS;
22
23 //010-00000000000000000000000000000,十进制值:1073,741,824
24 //此状态表示所有的任务已经被终止
25 private static final int TIDYING = 2 << COUNT_BITS;
26
27 //011-00000000000000000000000000000
28 //此状态表示已清理完现场
29 private static final int TERMINATED = 3 << COUNT_BITS;
30
31 // Packing and unpacking ctl
32 // 与运算,比如 001-00000000000000000000000100011,表示67个工作线程
33 // 掩码取反 111-00000000000000000000000000000,即得到最左边3位 001
34 // 表示线程处于stop 状态
35 private static int runStateOf(int c) { return c & ~CAPACITY; }
36 //同理掩码 000-11111111111111111111111111111,得到右边29位,即工作线程数
37 private static int workerCountOf(int c) { return c & CAPACITY; }
38 //把左边3位与右边29位按或运算,合并成一个值
39 private static int ctlOf(int rs, int wc) { return rs | wc; }
第一处说明,线程池状态用高3位表示,其中包括了符号位。五种状态的十进制值按小道大依次排序为:
RUNNING<SHUTDOWN<STOP<TIDYING<TERMINATED
这样设计的好处是可以通过比较值的大小来确定线程池的状态,例如程序中经常出现isRuning的判断。
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
我们都知道Exexutor 接口有且只有一个方法execute,通过参数传入待执行线程对象。下面分析ThreadPoolExecutor关于execute的实现。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//返回包含线程数及线程池状态的Integer类型数值
int c = ctl.get();
//如果工作线程数小于核心线程数,则创建线程任务并执行
if (workerCountOf(c) < corePoolSize) {
//addWorker 是另一个极为重要的方法,见下一段源码分析(第1处)
if (addWorker(command, true))
return;
//如果创建失败,防止外部已经在线程池中加入新任务,重新获取一下
c = ctl.get();
}
// 只有线程池处于RUNNING 状态,才执行后半句:置入队列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 如果线程池不是 RUNNING 状态 则将刚加入队列的任务移除
if (! isRunning(recheck) && remove(command))
reject(command);
//如果之前的线程已经被消费完,新建一个线程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//核心池和队列都已经满,尝试创建一个新线程
else if (!addWorker(command, false))
//如果addWorker返回false,即创建失败,则唤醒拒绝策略(第2处)
reject(command);
}
第2处 发生拒绝的原因有两个(1)线程池状态非Runing (2)等待队列已满。
下面继续分析addWorker 源码。
/**
* 根据当前线程池状态,检查是否可以添加新的任务线程,如果可以则创建并启动任务
* 如果一切正常则返回true。返回false 的可能如下:
* 1.线程池没有处于RUNNING状态
* 2.线程工程创建新的任务线程失败
* @param firstTask 外部启动线程池时需要构造的第一个线程,它是线程的母体
* @param core 新增工作线程时的判断指标,解释如下
* true:表示新增线程时,需要判断当前RUNNING 状态的线程是否少于corePoolsize
* false:表示新增线程时,需要判断当前RUNNING 状态的线程是否少于maxmemPoolsize
* @return
*/
private boolean addWorker(Runnable firstTask, boolean core) {
//不需要任务预定义的语法标签,响应下文的continue retry.快速退出多层嵌套循环(第1处)
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c); // 当前线程池状态
// Check if queue empty only if necessary.
// 这条语句等价:rs >= SHUTDOWN && (rs != SHUTDOWN || firstTask != null ||workQueue.isEmpty())
// 满足下列条件则直接返回false,线程创建失败:
// rs > SHUTDOWN:STOP || TIDYING || TERMINATED 此时不再接受新的任务,且所有任务执行结束
// rs = SHUTDOWN:firtTask != null 此时不再接受任务,但是仍然会执行队列中的任务
// rs = SHUTDOWN:firtTask == null见execute方法的addWorker(null, false),任务为null && 队列为空
// 最后一种情况也就是说SHUTDONW状态下,如果队列不为空还得接着往下执行,为什么?add一个null任务目的到底是什么?
// 看execute方法只有workCount==0的时候firstTask才会为null结合这里的条件就是线程池SHUTDOWN了不再接受新任务
// 但是此时队列不为空,那么还得创建线程把任务给执行完才行。
//第2处
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
// 走到这的情形:
// 1.线程池状态为RUNNING
// 2.SHUTDOWN状态,但队列中还有任务需要执行
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize)) // 最大线程数不能超过2^29,否则影响左边3位的线程池状态值
return false;
//(第3处)
if (compareAndIncrementWorkerCount(c)) //当前活动线程数+1(第3处) 原子操作递增workCount
break retry; //操作成功跳出重试循环
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs) //如果线程池的状态发生变化则重试(第5处)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
//开始创建工作线程
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//利用worker构造方法中的线程池工厂创建线程,并封装成工作线程worker对象
w = new Worker(firstTask);
//注意这是worker 中属性对象thread (第6处)
final Thread t = w.thread;
if (t != null) {
//对ThreadExecutor 的敏感操作时,都需要持有主锁,避免在添加和启动线程时被干扰。
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());
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;// 更新largestPoolSize
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 启动新添加的线程,这个线程首先执行firstTask,然后不停的从队列中取任务执行
// 当等待keepAlieTime还没有任务执行则该线程结束。见runWoker和getTask方法的代码。
if (workerAdded) {
t.start(); //最终执行的是ThreadPoolExecutor的runWoker方法 并非线程池execute 的command参数指向的线程
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}