Java多线程(二)——线程池

在多线程实践中,我们经常会用到线程池。池的概念没什么好废话的,各位应该比较熟悉,我们在这里简单总结下常用的内容。
我们可以使用ThreadPoolExecutor来创建线程池,它的构造方法如下:

1.corePoolSize,线程池的基本大小,当你提交任务到线程池时,线程池会创建一个线程来执行该任务。已经创建的线程数量达到corePoolSize之后就不再创建新线程了。

2.runnableTaskQueue,任务队列,用对保存需要执行的任务,阻塞队列,FIFO 队列 :LinkedBlockingQueue、ArrayListBlockingQueue(固定长度),优先级队列 :PriorityBlockingQueue。提供了阻塞的 take() 和 put() 方法:如果队列为空 take() 将阻塞,直到队列中有内容;如果队列满了 put() 将阻塞,直到队列有空闲位置。

3.maximumPoolSize,线程池最大大小,如果任务队列已经满了,则会继续创建新的线程继续执行任务。但是如果使用的任务队列是不是固定长度的,这个参数就是无效的。

4.RejectedExecutionHandler,拒绝策略,或者说饱和策略。任务队列满了之后,再进来的新任务必须做特殊处理,这里提供四种策略:
AbortPolicy:直接抛出异常。
CallerRunsPolicy:只用调用者所在线程来运行任务。
DiscardOldestPolicy:丢弃队列里最近的一个任务,并执行当前任务。
DiscardPolicy:不处理,丢弃掉。

实际工作中,我们也会自定义处理策略,需要实现RejectedExecutionHandler接口,如:持久化、日志等。

5.keepAliveTime,线程存活时间,线程池的压力不总是满负荷,keepAliveTime指定了空闲线程的存活时间。

线程池的工作流程如下图:

我们看一下它的execute方法代码:

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);
}

其实官方源码的注释已经写的非常详细了,我们在这里再做一些解读。首先,代码对你传递进来的Runnable做了空指针检查,这里没什么好说的。ctl,我们看代码可以发现它是一个private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0))原子整数。在这里,线程池的代码对它做了很奇妙的操作,用二进制运算对正数做了划分,它最高的三位维护了线程当前的状态,低29位维护了线程池中线程的数量,在这里可以不用过于计较。
首先判断当前线程池对象中已经开启的线程数量,如果小于corePoolSize,之前我们说过,这种情况下,会直接开启新的线程来处理提交的任务,也就是这里的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
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 {
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(w);
}
return workerStarted;
}

在addWorker的逻辑中,我们可以看到它首先检查了线程池的状态,然后进入循环重试的处理逻辑。如果线程池状态可以增加任务,我们可以看到它把参数的Runnable包装成了一个Worker,Worker是线程池的一个内部里,继承字AQS,并且实现了Runnable接口,它的构造方法如下:

1
2
3
4
5
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}

我们可以看到,它首先把AQS的状态设置为等待,然后把自己包装为线程对象。在把Worker增加到任务队列前,我们可以看到它尝试获取线程池内部的ReentrantLock,然后再次检查线程池状态,然后把worker加入到workers列表里。自此,线程数量少于corePoolSize的情况处理完毕,我们接下来看其他的分支。
if (isRunning(c) && workQueue.offer(command)) ,意思是,当线程数量已经达到了corePoolSize并且有任务在运行时,就使用offer方法把任务追加到woreQueue中。然后再次检查刚才加入的任务是否能执行,不能执行的话进行拒绝处理。如果线程池当前worker数量为0,则构造空worker拉起workerqueue的消费。
else if (!addWorker(command, false))到这一步表示,线程池不是running状态,而且已有线程已经全部占用,就尝试开启新线程,直到maxPoolSize,如果addWorker失败了,就拒绝任务。从excute的代码我们可以很清晰的看到我们之前对于线程池工作流程的描述。

我们看完添加任务的流程,我们再来看看任务是什么时候被消耗掉的。线程池的启动方法核心是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
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);
}
}

w.unlock,原因在于new Worker的时候,AQS的state = -1,-1状态下的AQS是不能中断的,这一处理主要作用在线程池的shutdown过程中,避免关闭线程池的时候把正常运行的worker中断。然后获取worker的锁,避免线程被其他地方关闭。然后就是正常的代理执行过程,执行完毕后释放锁,并且做对应的计数器处理。