源码分析之ThreadPoolExecutor

线程池在多线程编程的中可谓是个利器,使用线程池会大大提高多线程的效率。原因是使用线程池相对于new Thread有效避免了线程创建和销毁的开销。

Java中一般来说通过Executors来创建所需要的线程池,如:Executors.newFixedThreadPoolExecutors.newScheduledThreadPool等等。而其中一种主要的实现就是ThreadPoolExecutorThreadPoolExecutorJ.U.C包下的,当然也是出自大神Doug Lea之手。

ThreadPoolExecutor 结构

ThreadPoolExecutor类结构

如上图所示,ThreadPoolExecutor类大致结构如上图所示。Executor接口只有一个execute方法。ExecutorSerivce接口在Executor的基础上提供了对任务执行的管理,如:shutdown方法。AbstractExecutorService是针对其中invokeXXXsubmit做了默认的实现。真正的execute将由ThreadPoolExecutor自身实现。

ThreadPoolExecutor还有几个内部类:AbortPolicyDiscardPolicyDiscardOldestPolicyCallerRunsPolicy。这几个类的作用是当线程池承载很多任务之后,超过maximumPoolSize的数量,无法继续接受任务的时候,提供了不同的拒绝策略。

Worker类也是ThreadPoolExecutor的内部类,实现Runnable接口,提供了工作线程的实现。后面会详细分析这个类的实现。

ThreadPoolExecutor类提供了4个构造方法。其中最核心的构造方法如下所示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
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 超过corePoolSize数量的空闲线程最大存活时间
corePoolSize keepAliveTime 时间单位
workQueue 工作队列
ThreadFactory 线程工厂
RejectedExecutionHandler 拒绝策略,默认拒绝策略:AbortPolicy

ThreadPoolExecutor 类变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
// Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~CAPACITY; }
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

AtomicInteger ctl

ctl是一个AtomicInteger修饰的变量,用来保存2个变量:

  • workerCount: 线程池中活动线程的数量
  • runState:线程池的运行状态

workerCount

workerCount占用低29位存线程数,workerCount代表了线程池活动的线程数量,最小数量是0,最大的数量是(1 << COUNT_BITS) - 1 (536870911);

1
2
3
4
5
6
7
1 << COUNT_BITS :
00000000 00000000 00000000 00000001 --> 00100000 00000000 00000000 00000000
(1 << COUNT_BITS) - 1 :
00100000 00000000 00000000 00000000 --> 00011111 11111111 11111111 11111111
1
private static int workerCountOf(int c) { return c & CAPACITY; }

workerCountOf方法是用从ctl中解析出workerCount的值来。由于CAPACITY高3位是000ctl的值与CAPACITY&操作的时候,高3位将被舍弃;由于CAPACITY低29位是全是1&操作会保持原值,这样workerCount的值就从ctl中解析出来了。

runState

runState占用高3位存线程状态,共有5个值:

  • RUNNING(-536870912):接受新任务,并处理队列任务
  • SHUTDOWN(0): 不接受新任务,但会处理队列中的任务
  • STOP(536870912): 不接受新任务,不会处理队列任务,中断正在处理的任务
  • TIDYING(1073741824): 所有任务已结束,workerCount为0,线程过渡到TIDYING状态,将调用terminated()方法
  • TERMINATED(1610612736): terminated()方法已经完成
1
private static int runStateOf(int c) { return c & ~CAPACITY; }

runStateOf方法的作用是解析出runState的值。~CAPACITY将反转CAPACITY的值,也就是CAPACITY的高3位全部为1,低29位全部为0。如此ctl & ~CAPACITY 就会送ctl中解析出runState的值。

worker 类

线程池维护的是HashSet<Worker> workers,一个由Worker对象组成的HashSet

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
68
69
// 继承AQS,实现Runnable接口
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
// 处理任务的线程
final Thread thread;
// worker 传入的任务
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 创建一个任务线程
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.
protected boolean isHeldExclusively() {
return getState() != 0;
}
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}

Worker继承与AQS,主要想利用AQS独占所机制,来标示线程是否空闲。后续的shutdown方法在执行时候,判断线程是否空闲,tryLock来实现的。同样可以看到在Worker在初始化的时候将state设置为-1,这也是为了避免worker在执行前被中断。

Worker继承与Runnable,当worker启动时候就会调用run方法。Worker类的run方法调用了runWorker方法;

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
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
// 将state设置为0,允许中断
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// task为空,则调用getTask(),从workQueue取出新的task
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状态,要中断当前线程
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,循环操作
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}

execute 方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
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);
}

execute主要分3个步骤:

  • 当活动线程小于corePoolSize的值,将会调用addWorker创建新的任务线程;
  • 如果任务可以成功被加入至workQueue,这是将再次获取ctl的值做双重校验,如果线程池已经shutdown,将从workQueue移除并拒绝该任务。
  • 如果任务队列加入失败,则拒绝任务。

addWorker 方法

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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
// 线程池状态大于SHUTDOWN值将不再接受新的任务,创建线程。
// rs == SHUTDOWN or firstTask == null or workQueue.isEmpty() 都不再接受任务
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操作增加workerCount的值,增加成功跳出循环
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
}
}
// workerCount增加成功进入下面的逻辑
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
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();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 启动任务线程
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
// 任务线程启动失败调用addWorkerFailed,将线程从线程池移除,将workerCount -1
addWorkerFailed(w);
}
return workerStarted;
}
```
### shutdown 方法
```java
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// 设置线程池状态为:SHUTDOWN
advanceRunState(SHUTDOWN);
// 中断所有任务线程
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}

调用线程池shutdown方法,将不再接受新的任务。首先会调用checkShutdownAccess方法检验是否具有线程shutdown的权限,然后将线程池的状态设置为SHUTDOWN,最后中断空闲线程,这里的中断是用Thread.interrupt()实现的,所以不会影响正在执行的线程,正在执行的的任务将会继续执行。

shutdownNow 方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// 将线程池状态设置为:STOP
advanceRunState(STOP);
// 中断所有线程
interruptWorkers();
// 获取队列中尚未被执行的任务
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}

shutdownNow方法与shutdown方法类似,不同是shutdownNow会将线程池的状态设置STOP,在中断过程中,少了tryLock获取锁的操作,所以不管线程是否空闲都将被中断,中断所有线程,但也不会强行终止正在执行的线程。最后返回阻塞队列中没有被执行的任务list