谈谈ThreadPoolExecutor线程池

线程池,凡是学过java的同学都不陌生,一两行简单的代码就能实现并发编程。但java.util.concurrent.ThreadPoolExecutor的源码读起来却是很绕,今天,就让我们来深入了解一下线程池吧。

本文贴出的源码均是基于jdk1.7.0_79版本(jdk1.6版本源代码与该版本代码有比较大差别)。

线程池的构造方法

ThreadPoolExecutor提供了四个构造方法,不过前三个都会转向最后一个构造方法。

源代码1

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// 最后一个构造方法
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}

这里,可以先大概了解一下各个参数的含义。

corePoolSize:核心池大小,线程池是用来放线程的,就像澡堂是供人洗澡的(解释:过去北方人会到公共澡堂洗澡)。池的大小是有限的,如果池里线程数打满了,就需要把新来的任务安排到等待队列中。

maximumPoolSize:池的最大大小,可以理解为临时应急允许达到的最大数量,就像澡堂如果等待的人实在太多,就再开放一些淋浴位置。不过一般这个字段没用,我们使用线程池就是想有一个确定的最大线程运行数量,这个值一般和核心池大小值一致。

keepAliveTime:当线程池中的线程没有可以运行的新任务时,等待多久后自行结束,就像澡堂的淋浴喷头在等待几秒钟后如果没有人洗澡就会自动关闭。

workQueue:等待队列,当线程池运行的线程数量达到了核心池大小,新来的任务就会进到这里,就像澡堂里会有一个休息室。Executors.newFixedThreadPool(int)方法默认使用LinkedBlockingQueue<Runable>

threadFactory:构造Thread的工厂,要实现的方法是Thread newThread(Runnable r),就像澡堂中创建一个淋浴位置一样,可以给淋浴位置设置一下名称,组别之类的。

handler:当线程池线程数满了,等待队列也满了,也达到了临时应急设置的最大线程数量的时候,要执行的方法,就像澡堂洗澡的人满了,等待的人满了,澡堂该如何处理新进来要洗澡的人。

RejectedExecutionHandler接口的方法是rejectedExecution(Runnabler, ThreadPoolExecutor e),java默认实现了四种拒绝执行处理器,并默认使用AbortPolicy处理器。

  1. CallerRunsPolicy:如果线程池没有关闭,就执行这个任务。也就是说会在运行线程池提交任务方法的线程中运行线程。
  2. AbortPolicy:抛出一个RejectedExecutionException异常。
  3. DiscardPolicy:忽略这个任务,并且什么也不做。
  4. DiscardOldestPolicy:立刻取出等待队列中头一个任务(也就是等待时间最长的任务),然后再由线程池执行新任务。

向线程池提交任务

在第一节中,我提到了线程和任务,一定要区分这两个概念。使用者提交的是任务,可以在线程池中运行,或者被放到等待队列中。线程是线程池创建的,使用者的任务是被线程执行的。

我们来看看提交任务的源代码吧

源代码2

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
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);
}

翻译一下源代码中的注释:

  1. 如果线程池的线程数少于corePoolSize大小,尝试创建一个线程并执行提交的任务。addWorker方法会检查线程池的状态是不是可以提交任务,并且检查线程数量。通常它不会返回false。
  2. 如果任务可以加入到等待队列中,那么仍需检查我们是否应该添加一个线程,因为自提交任务到添加到队列中间有可能有线程已经退出。如果有必要,就从队列中删除该任务,并拒绝该任务的执行。或者如果有线程退出,则开始一个空任务新线程。
  3. 如果我们不能把任务加入到等待队列,则再次尝试开始一个带有该任务的新线程。如果失败,则说明等待队列满了,线程池数量等于允许的最大线程池数量,因此拒绝该任务的执行。

源代码2看似简单,其实还有些难以理解。我们根据注释的三部分来分别理解一下。

第一部分

先判断线程池的线程数量是不是小于设定的corePoolSize。当条件成立时,调用addWorker方法,返回结果为true就结束。

第二部分

说明线程池的线程数量大于等于corePoolSize,或者调用addWorker方法,返回结果为false。
这时,会先判断线程池是否还在运行,并且尝试把任务添加到等待队列中。

如果能加到等待队列中,会再次判断线程池是否还在运行。如果不在运行了,就移除添加到等待队列中的任务。并且调用rejectHandler拒绝该任务执行。

如果线程池还在运行,或者移除等待队列中的任务失败,则再判断线程池的线程数量是不是等于0,如果等于0,就创建一个空线程。

为什么要进行上边这个步骤?而不是直接添加到等待队列中就结束?原因是workQueue的offer方法执行是有时间的。

  1. 当offer成功后,可能已经过去了几十毫秒,这时线程池如果停止运行,应当把等待队列中的任务清除调并执行拒绝方法。
  2. 还有一种可能,offer过去了几十毫秒,线程池中的线程都已经运行完成结束了,没有线程能够从等待队列中取任务了,这个时候应该创建一个空线程。

第三部分

说明线程池的线程数量大于等于corePoolSize。或者调用addWorker方法,返回结果为false后,线程池停止运行了或者等待队列满了。

这是什么意思呢?线程池的线程数量大于等于corePoolSize了,等待队列满了,或者说线程池停止运行了。这时候能干啥呢?尝试创建一个带任务新线程,如果还失败,说明达到最大允许的最大线程池数量,只能拒绝方法运行了。

线程池创建线程执行任务

在源代码2中,出现了3次的addWorker方法是创建线程的关键。我们来看一下addWorker方法的源码。

源代码3

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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;
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
}
}

boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
final ReentrantLock mainLock = this.mainLock;
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int c = ctl.get();
int rs = runStateOf(c);

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;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

源代码3有点长,我们先来看上半部分for循环。

1
2
3
4
5
6
//Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;

注释中说了,在必要时检查等待队列是否为空。意思是,在运行状态为shutdown,firstTask为null,等待队列为空的时候,返回false。

1
2
3
4
5
6
7
8
9
10
11
12
for (;;) {
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
}

这段代码中,先会判断正在运行的线程数量有没有超过指定corePoolSize,注意,入参的第二个参数在这里用到,当core=true时,是和corePoolSize判断,core=false时,是和maximumPoolSize判断。

判断如果没超过,尝试使用cas更新线程池数量值。更新成功就开始真正的新增线程,否则就继续循环。

源代码3的下半部分比较简单,会先创建一个Worker,接着对代码段加锁。因为加锁会耗时,所以加完锁后再判断是否线程池停止运行。没有的话,将Worker添加到HashSet类型的workers中,解锁。解锁完成后,启动这个Worker,结束。

再看Worker

通读上边两节,我们成功地区分了线程和任务。也知道了线程被包裹在Worker中放到了HashSet中。那Worker到底是什么?Worker运行完后怎么从HashSet类型的workers中退出呢?来看源码吧!

源代码4

1
2
3
4
5
6
7
8
9
10
11
// 构造函数
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
// 实现了Runnable接口
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}

Worker是线程池中的内部类,实现了Runnable接口。根据源代码4可以看出,在源代码3中拿到的Worker.thread中包装的任务就是Worker自己,在源代码3中运行的也就是Worker的run方法。

源代码4中提到了该run方法实际上代理了实际任务。runWorker方法是运行实际任务的入口,来继续看源码吧!

源代码5

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
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
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);
}
}

略读源代码5,我们发现这里有一个while循环,尝试从Worker本身的firstTask和getTask()方法中获取可执行的任务。什么意思呢?也就是说当你在源代码2中调用addWorker时,第一个参数就是firstTask,运行Worker时,也就会先运行firstTask,运行完firstTask后才会调用getTask()方法从别的地方获取新任务。

其他地方需要注意的是,java.util.concurrent.ThreadPoolExecutor在这里给我们暴露出了两个方法,方便我们观察任务执行。他们分别是:

1
2
protected void beforeExecute(Thread t, Runnable r) { }
protected void afterExecute(Runnable r, Throwable t) { }

你可以自己继承线程池,重写这两个方法。线程池的作者给了一个例子,允许外部暂停线程池执行任务。

在方法processWorkerExit中,线程池的workers会移除当前Worker。

getTask()方法是我们最后要追寻的方法,下面是源码。

源代码6

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?

retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);

// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}

boolean timed; // Are workers subject to culling?

for (;;) {
int wc = workerCountOf(c);
timed = allowCoreThreadTimeOut || wc > corePoolSize;

if (wc <= maximumPoolSize && ! (timedOut && timed))
break;
if (compareAndDecrementWorkerCount(c))
return null;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}

try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}

其中我们看到,实际上会从线程池的等待队列workQueue中获取任务,如果线程池的线程数量超过corePoolSize,那么从等待队列中等待指定超时时间获取任务(poll),否则被等待队列调起(take)。这两个方法分别会让线程陷入timed_waiting状态和waiting状态。

为什么这么设计?原因是当线程池线程数量超过corePoolSize后,需要在尽量短的时间内销毁掉多余的线程,如果从等待队列中等待指定时间没有获取到任务,说明资源不紧张了,可以销毁线程了。

流程图

看了这么多是不是还有点头晕,这里我简单画了一个提交任务的流程图和runWorker运行任务的流程图,供大家参考理解。

upload successful

upload successful