ThreadPoolExecutor源码解析（二）
2024-08-29 16:52:47
1.ThreadPoolExcuter运行实例
首先我们先看如何新建一个ThreadPoolExecutor去运行线程。然后深入到源码中去看ThreadPoolExecutor里面使如何运作的。
public class Test {
public static void main(String[] args){
/**
* 新建一个线程池
* corePoolSize:2
* maximumPoolSize:10
* keepAliveTime:20
* unit:TimeUnit.SECONDS(秒)
* workQueue:new ArrayBlockingQueue(10)
* threadFactory:默认
* RejectedExecutionHandler默认
*/
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2,10,20, TimeUnit.SECONDS,new ArrayBlockingQueue(10));
//用execute添加一个线程
threadPool.execute(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
2.ThreadPoolExecute.execute方法
可以发现，其实使用线程池ThreadPoolExcuter 就是使用这个方法，然后我们看这个方法具体的代码。
/**
* 在后面执行给定任务。任务在一个新的线程中或一个存在的worker的线程池中执行。
* 如果一个线程不能提交到excution，可能是因为这个excutor已经shundown或者因为其容量已经是最大，
* 此时任务将会被RejectedExecutionHandler处理
*
*/
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
* 有以下3个步骤
*
* 1.如果少于corePoolSize的线程在运行，那么试着启动一个新线程，其中用给定指令作为first task。
* 这会调用addWorker去原子性得检查runState和workerCoune，因此可以防止错误报警，在错误报警不应该时通过返回false来添加线程
* 2.如果任务被成功排队，我们任然应该第二次检查是否添加一个新线程（因为可能存在在最后一次检查后挂掉的情况）
* 或者在进入这个方法期间线程池shutdown。所以我们再次检查状态，如果已关闭和有必要则退出队列，或者如果没有的话就开始一个新的线程。
* 3.如果我们无法将task入队，那么我们试图添加新线程。如果失败，那么知道我们shutdown或者是饱和的并拒绝task。
*/
int c = ctl.get();
//判断是否小于corePoolSize
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//如果pool在运行并且能提交到队列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//这里进行再次检查，如果线程池没在运行并且成功删除task后，使用拒绝策略拒绝该task
if (! isRunning(recheck) && remove(command))
reject(command);
//如果已经将task添加到队列中，而此时没有worker的话，那么新建一个worker。稍后这个空闲的worker就会自动去队列里面取任务来执行
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//如果无法提交那么按照拒绝策略拒绝task
else if (!addWorker(command, false))
reject(command);
}
线程池的ThreadPoolExcuter基础和 ThreadPoolExcuter Worker基本介绍在前一节已经有说过，可以点这里查看。可以看到这个方法的主要流程，其实都在注释里面说明了。可以发现里面主要调用了一个方法，addWorker()。 那么这个addWorker()又是什么东西呢。其实看方法名就很清楚了，就是新建一个Worker来执行你添加进来的task。
3.ThreadPoolExecute.addWorker()方法
/**
* 检查当前的线程池状态和容量，是否可以让一个新的worker加入。如果可以，worker计数将会被调整，并且
* 如果可能，一个新的woker将会被创建和开始，将它当作第一个任务来运行。当线程池是stopped或shutdown状态时，
* 将返回false。当线程工厂创建失败而返回null或者抛出exception（比如典型的OOM）时，它也会返回fails。
* firstTask：新线程应该第一个运行的任务。当线程数少于corePoolSize时或是队列满时，workers使用一个初始化的
* first task来创建，用来进行分流。初始化空闲线程通常使用prestartCoreThread。
* core:为true,如果使用有界的corePoolSize，否则时maxPoolSize
* @return true if successful
* 添加Worker
*/
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
//状态值
int rs = runStateOf(c); // Check if queue empty only if necessary.
//关于状态值的检测
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false; for (;;) {
int wc = workerCountOf(c);
//关于容量的检测
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//用到了原子CAS方法比较，使用CAS增加worker计数器成功，才能进入下一步
if (compareAndIncrementWorkerCount(c))
break retry;
//重新获取ctl
c = ctl.get(); // Re-read ctl
//这里表示执行到这里的时候线程池的运行状态改变，需要重新跳到retry处执行
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
} boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//使用firstTask初始化Worker，first可能为null，那么则表示该worker为空闲
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
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();
//largestPoolSize为跟踪的目前最大线程数，因为之前已经做过判断，所以不会越界问题
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
//workerAdded是在上面最后才设置的，确保这个变量能准确表示是否添加worker成功
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
//再次检查
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
addWorker本事只是为线程池添加一个ThreadPoolExcuter Worker，其本身所做的事情其实很简单，但难就难在要确保安全有效得添加一个ThreadPoolExcuter Worker。为此addWorker()方法做了很多额外的工作。比如判断线程池的运行状态，当前Worker数量是否已经饱和等等。可以发现在这个方法，或者说整个ThreadPoolExecutor中，很多时候都是使用双重检查的方式来对线程池状态进行检查。其实这都是为了效率，最简单不过直接使用Synchronized或ReentranLock进行同步，但这样效率会低很多，所以在这里，只有在万不得已的情况下，才会使用悲观的ReentranLock。
addWorker的最后直接调用了t.start，这里的t其实就是Worker它自己。接下来再看Worker是如何运行的。
4.ThreadPoolExecute.runWorker()方法
/**
* 主要的Worker运行的循环。重复得获取从任务队列中取出task并执行它。
*/ final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
//取出firstTask，再将worker中的值-设置为null
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//不断循环取出线程运行
while (task != null || (task = getTask()) != null) {
w.lock();
//锁住线程
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
//如果当前线程是stop，那么将确认其为interrupted
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//调用钩子
beforeExecute(wt, task);
Throwable thrown = null;
try {
//运行
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
从源码中可以看出，一个ThreadPoolExcuter Worker的工作其实就是不断使用getTask()方法从队列中获取新的任务来执行。值得一提的是，初始化参数里面的时间戳参数就是在这个方法里面运用的。在循环体中每次都使用锁以保证当前worker在运行task过程中不会被中断。同时运行时还会去调用两个内置的钩子：beforeExecute()和afterExecute（），这两个方法默认实现时空的。
同时在运行的循环中每次都关注着ThreadPoolExecutor的运行状态，当线程池处于中断状态时，循环Worker的当前线程也会中断。
总结：说到这里就差不多把线程池运行task的流程说完了,当然其中忽略了很多的细节。但总而言之，ThreadPoolExecutor其实就是对worker进行管理，然后使用这些worker来执行用户提交的task。对用户提交的task的数量也进行一定的控制管理，比如超过一定数量时放入一个任务队列中等等。然后对线程池规定一些状态量，根据这些状态量对线程池进行控制。
如果觉得对你有帮助，不如花0.5元请作者吃颗糖，让他甜一下吧~~
最新文章
- Gulp常用前端流程自动化配置
- scala中的集合框架
- ThreadLocal的理解
- char *p = "abcdefg"; p[0] = p[1]出错
- linux磁盘限额和进阶文件系统的管理 quota RAID LVM
- PHP用ajia代码写三级联动下拉
- QF——网络之JSON解析和XML解析
- MaskEdit组件的EditText属性和Text属性
- 安裝pycharm
- JavaScript进阶(九)JS实现本地文件上传至阿里云服务器
- Jquery浅克隆与深克隆
- laravel5.5 env
- MQ 简单的使用
- [字符串][NOIP2012]Vigenère密码
- BFS求解迷宫的最短路径问题
- 《杜增强讲Unity之Tanks坦克大战》11-游戏流程控制
- 华中科技大学 ubuntu14.04源
- filebeat向kafka中传输数据报WARN Failed to connect to broker DOMSDev07:9092: dial tcp: lookup DOMSDev07: getaddrinfow: No such host is known.解决方法
- xcopy命令的其他参数
- bzoj 3209 bzoj1799 数位dp
热门文章
- mysql 架构篇系列 1 复制原理和复制架构
- 近期遇到的计(算)算(法)题及解(JavaScript)
- TCPWrap的使用配置
- Spring-WebSocket 教程
- Nodejs sublime text 3安装与配置
- Pandas数据去重和对重复数据分类、求和，得到未重复和重复（求和后）的数据
- JS_正则表达式_使用字符串创建的正则表达式_反斜杠也需要添加转义符
- 设置TabWidget的样式的方法、关联Fragment与tabwidget的方法、点击tab显示相应Fragment方法
- WPF BackGroundWord 异步加载更新进度条示例
- ABP Changing Httpcode status