阅读任何源码,我们都应该带着几个问题去阅读,从源码中找出这些问题的答案,这样才能彻底搞明白某个知识点。
下面我们就带着这样几个问题,一起看一下ThreadPoolExecutor的源码
- 为什么要用线程池
- 为什么不推荐使用juc直接创建的线程池
- 线程池的几个核心参数
- 线程池是什么时候创建线程的?
- 线程池是如何重复利用线程的?
- 任务提交的顺序和执行的顺序是一样的吗?
1、为什么要使用线程池
这个其实可以写一个简单的程序去跑一下,比如使用线程池去跑1000个task和开1000个线程去跑这1000个task,线程池的效率会高出很多倍,原因是线程池能够重复利用线程,没有创建和销毁线程的开销。
其实池化的技术在很多地方都会用到比如数据库的连接池,字符串常量池,netty的对象池等等
2、为什么不推荐使用juc直接创建线程池的方式
我们找两个Exectors创建线程池的源码
a、newCachedTreadPool的源码
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
可以看到这里的最大线程数是0xffff个,这个最大线程数在大并发提交任务的情况下会创建大量线程,会导致CPU100%
b、newSingleThreadExecutor
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory));
}
//看一下LinkedBlockingQueue的实现
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
会初始化一个容量为0xffff的队列,由于这个队列太大,如果我们提交的任务数很多并且自定义的线程里的对象又很大的话,就很容易发生oom的问题。
从这个工具类创建线程的参数我们可以看到底层调用的都是ThreadPoolExecutor的构造方法,所以我们建议根据具体业务的规模设置合适的线程池参数。
new ThreadPoolExecutor()
3、线程池的几个核心参数
ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue)
corePoolSize:最大线程数
maximumPoolSize:最大线程数
keepAiveTime:线程存活时间
TimeUnit:存活时间的参数
BlockingQueue:线程池的任务队列
task投递到线程池中的整个过程如下
看一下具体的代码
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//这里判断线程数量是否小于corePoolSize,如过小于corePoolSize则直接创建worker线程
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//到这个if说明worker线程数量大于了corePoolSize了,这里直接添加到任务队列
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);
}
//到这个if说明任务加入队列失败,队列满了,则再创建线程worker线程,如果创建失败则执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}
看到这里从宏观上我们就能看到整个task投递到线程池的一个过程,其实这里最主要的方法是addWorker,其实addworker里才是线程池的精华,里面有如何创建线程及start的逻辑,如何回收过期的线程,如何重复利用创建的线程去运行task的逻辑
4、线程池是什么时候创建线程的
线程池的线程不是线程池创建的时候创建的,线程池的线程是在调用addWorker方法,并且addWorker执行成功才会创建
下面我们一起分析一下addworker代码,这个方法很长,其实要看明白这个方法我们要先看一下Worker这个类,可以看到这个类实际上实现了Runnable接口,其实线程池中运行的runnable任务都会被包装成Worker对象
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
//当前worker会绑定一个线程
final Thread thread;
//当前worker处理的第一个任务
Runnable firstTask;
volatile long completedTasks;
//创建worker时就会生成 一个线程及赋值firstTask,
//注意线程池的线程就是在这里被创建的
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
//注意看这里,这个线程创建的时候传的是this,也就意味着一会this.thread.start时,
//执行的是worker对象的run方法
//这个大家想一下,回味一下
this.thread = getThreadFactory().newThread(this);
}
}
下面在具体分析一下addWorker是如何start线程及重复利用线程运行任务的
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
//获取worker数量
int c = ctl.get();
//获取线程池状态
int rs = runStateOf(c);
// 如果线程池状态为SHUTDOWN就不接收任务了直接returnfalse
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
//这里会判断一下worker数量
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
//如果代码运行到了这里,就表示线程池状态正常,任务达到了创建worker线程的条件
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//这里会创建一个worker对象,
//结合上面的代码,worker对象中会包含一个线程和一个firstTask
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
//下面这一部分是用来判断需不需要将worker线程进行缓存,给其他的任务使用
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//注意看这个workers对象,这是一个hashset,用来缓存创建好的worker对象
//注意在强调一下这个worker对象包含一个Thread引用及Runnable引用
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
//缓存好worker线程后,这里会执行线程的start逻辑
//注意线程池的任务就是在这里开启start使任务真正进入runnable状态的
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
下面我们看一下start的具体逻辑:
刚刚我们已经看到了,thread是worker对象的一个属性,实际上创建线程时,传的参数就是worker对象本身,所以线程start执行的逻辑就是worker对象的run方法
//worker的run方法很简单,下面我们看一下runWorker方法
public void run() {
runWorker(this);
}
看完这个我们就基本上看完了线程池的核心逻辑了,但是还有一些细节没有仔细看,后面我会提到,留给大家思考
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
//获取一下fisrtTask
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//这个task什么时候不为空?
//注意这里会有一个隐含的问题,我们提交的任务存放的顺序是 核心worker->队列->最大线程worker
//这里的getTask是从队列里获取的任务
//这里task的执行顺序就变成了,核心worker->最大线程worker->队列
//不知道大家有没有get到我的点,可以做一个实验就是为task编一个号,比如安1-10的顺序提交任务,最终执行的结果却是 1,2,3,4,8,9,10,5,6,7这种顺序,这里就是出现这种提交顺序和执行顺序不一样的原理
while (task != null || (task = getTask()) != null) {
w.lock();
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方法,到此线程池的创建到运行任务就看结束了
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 {
//这里会销毁线程,在队列为空的时候,会销毁线程,让线程数量停留在corePoolsize范围内,什么时候会执行到这,当task为空的时候,什么时候task为空,看getTask的逻辑,getTask会判断队列是否为空及活跃线程的时间来返回task的值,这里就不细讲了
processWorkerExit(w, completedAbruptly);
}
}
5、线程池是如何重复利用线程的
看完上面的分析,其实就能回答这个问题了,当任务提交时会创建worker对象,这个对象里会有绑定一个线程,同时会将worker对象放入workers Set中,创建成功后,会立马调用worker.thread.start方法启动线程,这个线程的run方法进行了包装,首先判断fisrtTask是否为空如果不为空则直接运行,否则会while循环拿缓存队列中的任务,知道缓存队列为空,或者空闲线程超过了keepalive时间就会销毁线程,以保证线程维持在corePoolSize的大小
6、任务提交的顺序和执行的顺序是一样的吗?
不一样,提交顺序是 核心worker线程->队列->非核心worker线程,
但是执行的顺序却是核心worker任务->非核心worker任务->队列任务
如果能理解这个,线程池就真的理解的差不多了
7、其他
上面几个问题有些是我在看源码时的困惑点,有些是我看完源码之后的一些想法,除了这些问题外,线程池还有很多精妙的地方比如,
a、线程池的状态和核心线程数其实是用一个4个字节int表示的,为什么要这么表示
b、线程池中使用到的设计模式有哪些
c、线程池本身就是并发场景下提交任务的,那它自己的安全性是如何保证的,execute方法是如何保证安全性的