线程池_03_源码分析
-
关键变量介绍：
-
线程的状态
-
5种
- RUNNABLE：运行状态，接受新任务，持续处理任务队列里的任务
- SHUTDOWN：不再接受新任务，但要处理任务队列里的任务
- STOP：不接受新任务，不再处理任务队列里的任务，中断正在进行中的任务
- TIDYING：表示线程池正在停止运作，中止所有任务，销毁所有工作线程
- TERMINATED：表示线程池已停止运作，所有工作线程已被销毁，所有任务已被清空或执行完毕
-
状态装换
-
细节
- 线程的状态使用ctl表示，默认为 running
- private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
-
ctl类型为AtomicInteger，那用一个基础如何表示以上五种状态以及线程池工作线程数量呢？int型变量占用4字节，共32位，因此采用位表示，可以解决上述问题。5种状态使用5种数值进行表示，需要占用3位，余下的29位就可以用来表示线程数。因此，高三位表示进程状态，低29位为线程数量，代码如下：
- private staticfinal int COUNT_BITS = Integer.SIZE - 3; // 值为29
- private staticfinal int CAPACITY = (1 <<COUNT_BITS) - 1; //高三位全为0，低29位全为1，因此线程数量的表示范围为 0 ～ 2^29
-
因为ctl分位来表示状态和数量，下面几个状态仅看有效位的值
- private staticfinal int RUNNING = -1 <<COUNT_BITS; // 有效值为 111
- private staticfinal int SHUTDOWN = 0 << COUNT_BITS; // 有效值为 000
- private staticfinal int STOP = 1 << COUNT_BITS; // 有效值为 001
- private staticfinal int TIDYING = 2 << COUNT_BITS; // 有效值为 010
- private staticfinal int TERMINATED = 3 <<COUNT_BITS; // 有效值为 011
- 采用int分位表示线程池状态和线程数量; 并且提供runStateOf(): 获取线程池状态 和 workerCountOf(): 获取工作线程数量两个方法，均为二进制操作。
-
5种
-
工作线程(Worker)
- 线程池中的工作线程以Worker作为体现，真正工作的线程为Worker的成员变量，Worker即是Runnable，又是同步器。Worker从工作队列中取出任务来执行，并能通过Worker控制任务状态。
-
线程的状态
-
执行任务，源码分析：
-
1. execute() --提交线程任务;
-
2. addWorker() -- 通过添加核心和非核心线程来执行任务
-
private boolean addWorker(Runnable firstTask, boolean core) {
- int c = this.ctl.get(); // 获取当前ctl值
- label253:
- while(!runStateAtLeast(c, 0) || !runStateAtLeast(c, 536870912) && firstTask == null && !this.workQueue.isEmpty()) {
-
// workerCountOf(c)-获取线程数 < core ? this.corePoolSize : this.maximumPoolSize 如果为true,核心线程;否则,非核心线程
- while(workerCountOf(c) < ((core ? this.corePoolSize : this.maximumPoolSize) & 536870911)) {
- // CAS操作增加线程数，跳出循环
-
if (this.compareAndIncrementWorkerCount(c)) {
- boolean workerStarted = false;
- boolean workerAdded = false;
- ThreadPoolExecutor.Worker w = null;
-
try {
- w = new ThreadPoolExecutor.Worker(firstTask);
- Thread t = w.thread;
-
if (t != null) {
- ReentrantLock mainLock = this.mainLock;
- mainLock.lock();
-
try {
-
int c = this.ctl.get();
-
if (isRunning(c) || runStateLessThan(c, 536870912) && firstTask == null) {
-
if (t.isAlive()) {
- throw new IllegalThreadStateException();
- }
- this.workers.add(w);
- int s = this.workers.size();
-
if (s > this.largestPoolSize) {
- this.largestPoolSize = s;
- }
- workerAdded = true;
-
if (t.isAlive()) {
- }
-
if (isRunning(c) || runStateLessThan(c, 536870912) && firstTask == null) {
-
} finally {
- mainLock.unlock();
- }
-
if (workerAdded) {
- t.start();
- workerStarted = true;
- }
-
int c = this.ctl.get();
- }
-
} finally {
-
if (!workerStarted) {
- this.addWorkerFailed(w);
- }
-
if (!workerStarted) {
- }
- return workerStarted;
- }
- c = this.ctl.get();
- // 上面的CAS操作没成功，检查线程池状态与开始是否一致；如果一致，继续执行此for循环，否则重新执行retry代码块；自旋以期CAS成功，后续才能添加线程
-
if (runStateAtLeast(c, 0)) {
- continue label253;
- }
- }
- return false;
-
}
- return false;
- }
-
addWorkerFailed(w)
- 首先是将Worker移除，然后通过CAS操作更新ctl，最后调用tryTerminate()操作尝试中止线程池。
-
private boolean addWorker(Runnable firstTask, boolean core) {
-
4. runWorker()
- 线程首个任务为firstTask，之后通过getTask()就从阻塞队列里任务。线程池提供了beforeExecute()和afterExecute()通知子类任务执行前后的回调，让子类有时机能执行自己的事情。如果线程池已没有任务了，工作线程达到了可退出的状态，则将线程退出。
-
线程池执行的任务的线程，也就是Workder里的Thread。因此在addWorker()中执行new ThreadPoolExecutor.Worker(firstTask)后;执行的是Worker.run()，run()则调用了ThreadPoolExecutor.runWorker()
-
5. getTask()
- 线程池里的线程从阻塞队列里拿任务，如果存在非核心线程，假设阻塞队列里没有任务，那么非核心线程也要在等到keepAliveTime时间后才会释放。
-
如果当前仅有核心线程存在，如果允许释放核心线程的话，也就和非核线程的处理方式一样，反之，则通过take()一直阻塞直到拿到任务，这也就是线程池里的核心线程为什么不死的原因。
-
6. processWorkerExit()
- 在线程没有拿到任务后，退出线程
-
释放工作线程也并没有区分核心与非核心，也是随机进行的。所谓随机，就是在前面所说的区间范围内，根据释放策略，哪个线程先达到获取不到任务的状态，就释放哪个线程。
-
tryTerminate()
- 发现可以中止线程池时，中止，并调用terminated()进行通知。如果线程池处于RUNNABLE状态，什么也不做，否则尝试中断一个线程。中断线程通过interruptIdleWorker()完成。
-
1. execute() --提交线程任务;