并发学习(七) Java线程池的分析与使用

Java线程池的分析与使用

使用线程池的好处

  • 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
  • 提高响应速度。当任务到达时,任务可以不需要的等到线程创建就能立即执行。
  • 提高线程的可管理性。线程是稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一的分配,调优和监控。

继承体系

threadpool.png
如上图所示,最顶层是一个 Excutor 接口,它之声明excute方法。而下层的ExecutorService接口继承了上层的Execute接口,声明了shutdown、submit、invokeAll、invokeAny 等方法。再下层的ScheduledExecutorService接口声明了chedule和scheduleAtFixedRate等定时任务相关的方法。线程池的核心实现是在 ThreadPoolExecutor 类中,调用Excutors的newFixedThreadPool、newSingleThreadExecutor和newCachedThreadPool等方法创建线程池均是 ThreadPoolExecutor 类型。

创建线程池

1
new ThreadPoolExecutor(corePoolSize, maximumPoolSize,keepAliveTime, milliseconds,runnableTaskQueue, threadFactory,handler);

当提交一个任务到线程池时,线程池会创建一个线程来执行任务,即使其它空闲的基本线程能够执行新任务也会创建线程,等到需要执行的任务数大于线程池基本大小时就不再创建。

向线程池提交任务

我们可以使用execute提交的任务,但是execute方法没有返回值,所以无法判断任务知否被线程池执行成功。通过以下代码可知execute方法输入的任务是一个Runnable类的实例。

1
2
3
4
5
6
7
8
9
10
threadsPool.execute(new Runnable() {
@Override

public void run() {

// TODO Auto-generated method stub

}

});

我们也可以使用submit 方法来提交任务,它会返回一个future,那么我们可以通过这个future来判断任务是否执行成功,通过future的get方法来获取返回值,get方法会阻塞住直到任务完成,而使用get(long timeout, TimeUnit unit)方法则会阻塞一段时间后立即返回,这时有可能任务没有执行完。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
try {

Object s = future.get();

} catch (InterruptedException e) {

// 处理中断异常

} catch (ExecutionException e) {

// 处理无法执行任务异常

} finally {

// 关闭线程池

executor.shutdown();

}

关闭线程池

可以通过调用线程池的shutdown或shutdownNow方法来关闭线程池,
shutdown的原理是只是将线程池的状态设置成SHUTDOWN状态,然后中断所有没有正在执行任务的线程。

shutdownNow的原理是遍历线程池中的工作线程,然后逐个调用线程的interrupt方法来中断线程,所以无法响应中断的任务可能永远无法终止。shutdownNow会首先将线程池的状态设置成STOP,然后尝试停止所有的正在执行或暂停任务的线程,并返回等待执行任务的列表。

只要调用了这两个关闭方法的其中一个,isShutdown方法就会返回true。当所有的任务都已关闭后,才表示线程池关闭成功,这时调用isTerminaed方法会返回true。至于我们应该调用哪一种方法来关闭线程池,应该由提交到线程池的任务特性决定,通常调用shutdown来关闭线程池,如果任务不一定要执行完,则可以调用shutdownNow。

原理分析

在了解线程池的使用后我们来对线程池核心参数进行分析。

构造方法核心参数分析

1
2
3
4
5
6
7
8
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
参数 说明
corePoolSize 核心线程数。当线程小于该值时,线程会优先创建新线程来执行新任务
maximumPoolSize 线程池所能维护的最大线程数
keepAliveTime 空闲线程的存活时间
workQueue 任务序列,用于缓存未执行的任务
threadFactory 线程工厂。可通过工厂为新建的线程设置更有意义的名字
handler 饱和策略。当线程池和任务队列均处于饱和状态时,使用拒绝策略处理新任务。默认是AbortPolicy,即直接抛出异常

线程池实现分析

5679451-2c3d68cec695c1ac.png

  1. 如果当前运行的线程少于corePoolSize,则创建新线程来执行任务(这一步需获取全局锁)。
  2. 如果运行的线程等于或多于corePoolSize,则将任务加入BlockingQueue。
  3. 如果队列已满,则创建新的线程来处理任务(这一步需获取全局锁)。
  4. 如果创建新线程将使当前运行的线程超出maximumPoolSize,任务将被拒绝,并使用相关饱和策略进行处理,默认是直接抛出异常。

简化规则:

序号 条件 动作
1 线程数<corePoolSize 创建新线程
2 线程数 ≥ corePoolSize,且 workQueue 未满 缓存新任务
3 corePoolSize ≤ 线程数 < maximumPoolSize,且 workQueue 已满 创建新线程
4 线程数 ≥ maximumPoolSize,且 workQueue 已满 使用拒绝策略处理

资源回收

考虑到系统资源是有限的,对于线程池超出 corePoolSize 数量的空闲线程应进行回收操作。进行此操作存在一个问题,即回收时机。目前的实现方式是当线程空闲时间超过 keepAliveTime 后,进行回收。除了核心线程数之外的线程可以进行回收,核心线程内的空闲线程也可以进行回收。回收的前提是allowCoreThreadTimeOut属性被设置为 true,通过public void allowCoreThreadTimeOut(boolean) 方法可以设置属性值。

排队策略

当线程数量大于等于 corePoolSize,workQueue 未满时,则缓存新任务。这里要考虑使用什么类型的容器缓存新任务,通过 JDK 文档介绍,我们可知道有3中类型的容器可供使用,分别是同步队列,有界队列和无界队列。对于有优先级的任务,这里还可以增加优先级队列。以上所介绍的4中类型的队列,对应的实现类如下:

实现类 类型 说明
SynchronousQueue 同步队列 该队列不存储元素,每个插入操作必须等待另一个线程调用移除操作,否则插入操作会一直阻塞
ArrayBlockingQueue 有界队列 基于数组的阻塞队列,按照 FIFO 原则对元素进行排序
LinkedBlockingQueue 无界队列 基于链表的阻塞队列,按照 FIFO 原则对元素进行排序
PriorityBlockingQueue 优先级队列 具有优先级的阻塞队列

饱和策略

线程数量大于等于 maximumPoolSize,且 workQueue 已满,则使用拒绝策略处理新任务。Java 线程池提供了4中拒绝策略实现类,如下:

实现类 说明
AbortPolicy 丢弃新任务,并抛出 RejectedExecutionException
DiscardPolicy 不做任何操作,直接丢弃新任务
DiscardOldestPolicy 丢弃队列队首的元素,并执行新任务
CallerRunsPolicy 由调用线程执行新任务

以上4个拒绝策略中,AbortPolicy 是线程池实现类所使用的策略。我们也可以通过方法public void setRejectedExecutionHandler(RejectedExecutionHandler)修改线程池决绝策略。

重要操作

线程的创建和复用

在线程池的实现上,线程的创建是通过线程工厂接口ThreadFactory的实现类来完成的。默认情况下,线程池使用Executors.defaultThreadFactory()方法返回的线程工厂实现类。

在线程池中,线程的复用是线程池的关键所在。这就要求线程在执行完一个任务后,不能立即退出。对应到具体实现上,工作线程在执行完一个任务后,会再次到任务队列获取新的任务。如果任务队列中没有任务,且 keepAliveTime 也未被设置,工作线程则会被一致阻塞下去。通过这种方式即可实现线程复用。

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
+----ThreadPoolExecutor.Worker.java
Worker(Runnable firstTask) {
setState(-1);
this.firstTask = firstTask;
// 调用线程工厂创建线程
this.thread = getThreadFactory().newThread(this);
}

// Worker 实现了 Runnable 接口
public void run() {
runWorker(this);
}

+----ThreadPoolExecutor.java
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock();
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);
}
}

提交任务

在之前我们分析过线程池提交任务的过程(三个判断),挺好理解,下面看看代码实现

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
96
97
98
99
100
101
102
103
104
105
106
+---- AbstractExecutorService.java
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
// 创建任务
RunnableFuture<Void> ftask = newTaskFor(task, null);
// 提交任务
execute(ftask);
return ftask;
}

+---- ThreadPoolExecutor.java
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();
}

// 缓存任务,如果队列已满,则 offer 方法返回 false。否则,offer 返回 true
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);
}

// 添加工作者对象,并在 addWorker 方法中检测线程数是否小于最大线程数
else if (!addWorker(command, false))
// 线程数 >= 最大线程数,使用拒绝策略处理任务
reject(command);
}

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 {
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// 将 worker 对象添加到 workers 集合中
workers.add(w);
int s = workers.size();
// 更新 largestPoolSize 属性
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 开始执行任务
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

几种线程池

一般情况下,我们并不直接使用 ThreadPoolExecutor 类创建线程池,而是通过 Executors 工具类去构建线程池。通过 Executors 工具类,我们可以构造5中不同的线程池。下面通过一个表格简单介绍一下几种线程池,如下:

静态构造方法 说明
newFixedThreadPool(int nThreads) 构建包含固定线程数的线程池,默认情况下,空闲线程不会被回收
newCachedThreadPool() 构建线程数不定的线程池,线程数量随任务量变动,空闲线程存活时间超过60秒后会被回收
newSingleThreadExecutor() 构建线程数为1的线程池,等价于 newFixedThreadPool(1) 所构造出的线程池
newScheduledThreadPool(int corePoolSize) 构建核心线程数为 corePoolSize,可执行定时任务的线程池
newSingleThreadScheduledExecutor() 等价于 newScheduledThreadPool(1)

参考资料