线程池原理与实战
Java线程的创建非常昂贵，需要JVM和OS（操作系统）配合完成大量的工作：
- 必须为线程堆栈分配和初始化大量内存块，其中包含至少1MB的栈内存。
- 需要进行系统调用，以便在OS（操作系统）中创建和注册本地线程。
Java高并发应用频繁创建和销毁线程的操作是非常低效的，而且是不被编程规范所允许的。如何降低Java线程的创建成本？必须使用到线程池。线程池主要解决了以下两个问题：
- 提升性能：线程池能独立负责线程的创建、维护和分配。在执行大量异步任务时，可以不需要自己创建线程，而是将任务交给线程池去调度。线程池能尽可能使用空闲的线程去执行异步任务，最大限度地对已经创建的线程进行复用，使得性能提升明显。
- 线程管理：每个Java线程池会保持一些基本的线程统计信息，例如完成的任务数量、空闲时间等，以便对线程进行有效管理，使得能对所接收到的异步任务进行高效调度。
说明在主要大厂的编程规范中，不允许在应用中自行显式地创建线程，线程必须通过线程池提供。由于创建和销毁线程需要时间以及系统资源开销，使用线程池的好处是减少这些开销，解决资源不足的问题。
1. JUC的线程池架构
在多线程编程中，任务都是一些抽象且离散的工作单元，而线程是使任务异步执行的基本机制。随着应用的扩张，线程和任务管理也变得非常复杂。为了简化这些复杂的线程管理模式，我们需要一个“管理者”来统一管理线程及任务分配，这就是线程池。在JUC中有关线程池的类与接口的架构图大致如图1-15所示。
JUC就是java.util.concurrent工具包的简称，该工具包是从JDK 1.5开始加入JDK的，是用于完成高并发、处理多线程的一个工具包。
1. Executor
Executor是Java异步目标任务的“执行者”接口，其目标是执行目标任务。“执行者”Executor提供了execute()接口来执行已提交的Runnable执行目标实例。Executor作为执行者的角色，其目的是提供一种将“任务提交者”与“任务执行者”分离开来的机制。它只包含一个函数式方法：
2.ExecutorService
ExecutorService继承于Executor。它是Java异步目标任务的“执行者服务接”口，对外提供异步任务的接收服务。ExecutorService提供了“接收异步任务并转交给执行者”的方法，如submit系列方法、invoke系列方法等，具体如下：
3.AbstractExecutorService
AbstractExecutorService是一个抽象类，它实现了ExecutorService接口。AbstractExecutorService存在的目的是为ExecutorService中的接口提供默认实现。
4.ThreadPoolExecutor
ThreadPoolExecutor就是大名鼎鼎的“线程池”实现类，它继承于AbstractExecutorService抽象类。
ThreadPoolExecutor是JUC线程池的核心实现类。线程的创建和终止需要很大的开销，线程池中预先提供了指定数量的可重用线程，所以使用线程池会节省系统资源，并且每个线程池都维护了一些基础的数据统计，方便线程的管理和监控。
5.ScheduledExecutorService
ScheduledExecutorService是一个接口，它继承于ExecutorService。它是一个可以完成“延时”和“周期性”任务的调度线程池接口，其功能和Timer/TimerTask类似。
6.ScheduledThreadPoolExecutor
ScheduledThreadPoolExecutor继承于ThreadPoolExecutor，它提供了ScheduledExecutorService线程池接口中“延时执行”和“周期执行”等抽象调度方法的具体实现。
ScheduledThreadPoolExecutor类似于Timer，但是在高并发程序中，ScheduledThreadPoolExecutor的性能要优于Timer。
7.Executors
Executors是一个静态工厂类，它通过静态工厂方法返回ExecutorService、ScheduledExecutorService等线程池示例对象，这些静态工厂方法可以理解为一些快捷的创建线程池的方法。
2. 线程池的标准创建方式
大部分企业的开发规范都会禁止使用快捷线程池（具体原因稍后介绍），要求通过标准构造器ThreadPoolExecutor去构造工作线程池。Executors工厂类中创建线程池的快捷工厂方法实际上是调用ThreadPoolExecutor（定时任务使用ScheduledThreadPoolExecutor）线程池的构造方法完成的。ThreadPoolExecutor构造方法有多个重载版本，其中一个比较重要的构造器如下：
1.核心和最大线程数量
参数corePoolSize用于设置核心（Core）线程池数量，参数maximumPoolSize用于设置最大线程数量。线程池执行器将会根据corePoolSize和maximumPoolSize自动维护线程池中的工作线程，大致规则为：
- 当在线程池接收到新任务，并且当前工作线程数少于corePoolSize时，即使其他工作线程处于空闲状态，也会创建一个新线程来处理该请求，直到线程数达到corePoolSize。
- 如果当前工作线程数多于corePoolSize数量，但小于maximumPoolSize数量，那么仅当任务排队队列已满时才会创建新线程。通过设置corePoolSize和maximumPoolSize相同，可以创建一个固定大小的线程池。
- 当maximumPoolSize被设置为无界值（如Integer.MAX_VALUE）时，线程池可以接收任意数量的并发任务。
- corePoolSize和maximumPoolSize不仅能在线程池构造时设置，也可以调用setCorePoolSize()和setMaximumPoolSize()两个方法进行动态更改。
2.BlockingQueue
BlockingQueue（阻塞队列）的实例用于暂时接收到的异步任务，如果线程池的核心线程都在忙，那么所接收到的目标任务缓存在阻塞队列中。
3.keepAliveTime
线程构造器的keepAliveTime（空闲线程存活时间）参数用于设置池内线程最大Idle（空闲）时长（或者说保活时长），如果超过这个时间，默认情况下Idle、非Core线程会被回收。
如果池在使用过程中提交任务的频率变高，也可以调用方法setKeepAliveTime(long，TimeUnit)进行线程存活时间的动态调整，可以将时长延长。如果需要防止Idle线程被终止，可以将Idle时间设置为无限大，具体如下：
默认情况下，Idle超时策略仅适用于存在超过corePoolSize线程的情况。但若调用了allowCoreThreadTimeOut(boolean)方法，并且传入了参数true，则keepAliveTime参数所设置的Idle超时策略也将被应用于核心线程。
3.向线程池提交任务的两种方式
向线程池提交任务的两种方式大致如下：
1
2
3
4
5
6
7
|
void execute（Runnable command)
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Feture<?> submit(Runable task);
|
以上的submit()和execute()两类方法的区别在哪里呢？大致有以下三点：
二者所接收的参数不一样
Execute()方法只能接收Runnable类型的参数，而submit()方法可以接收Callable、Runnable两种类型的参数。
Callable类型的任务是可以返回执行结果的，而Runnable类型的任务不可以返回执行结果。Callable是JDK 1.5加入的执行目标接口，作为Runnable的一种补充，允许有返回值，允许抛出异常。
Runnable和Callable的主要区别为：Callable允许有返回值，Runnable不允许有返回值；Runnable不允许抛出异常，Callable允许抛出异常。
submit()提交任务后会有返回值，而execute()没有
execute()方法主要用于启动任务的执行，而任务的执行结果和可能的异常调用者并不关心。submit()方法也用于启动任务的执行，但是启动之后会返回Future对象，代表一个异步执行实例，可以通过该异步执行实例去获取结果。
submit()方便Exception处理
execute()方法在启动任务执行后，任务执行过程中可能发生的异常调用者并不关心。而通过submit()方法返回的Future对象（异步执行实例），可以进行异步执行过程中的异常捕获。
1.通过submit()返回的Future对象获取结果
submit()方法自身并不会传递结果，而是返回一个Future异步执行实例，处理过程的结果被包装到Future实例中，调用者可以通过Future.get()方法获取异步执行的结果。通过submit()返回的Future对象获取异步执行结果，演示代码如下：
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
|
@Test
public void testSubmit2() {
ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
Future<Integer> future = pool.schedule(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
return RandomUtil.randInRange(200, 300);
}
}, 100, TimeUnit.MILLISECONDS);
try {
Integer result = future.get();
Print.tco("异步执行的结果是:" + result);
} catch (InterruptedException e) {
Print.tco("异步调用被中断");
e.printStackTrace();
} catch (ExecutionException e) {
Print.tco("异步调用过程中，发生了异常");
e.printStackTrace();
}
sleepSeconds(10);
pool.shutdown();
}
|
2.通过submit()返回的Future对象捕获异常
submit()方法自身并不会传递异常，处理过程中的异常都被包装到Future实例中，调用者在调用Future.get()方法获取执行结果时，可以捕获异步执行过程中抛出的受检异常和运行时异常，并进行对应的业务处理。演示代码如下：
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
|
@Test
public void testSubmit() {
ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
pool.execute(new TargetTaskWithError());
Future future = pool.submit(new TargetTaskWithError());
try {
if (future.get() == null) {
Print.tco("任务完成");
}
} catch (Exception e) {
Print.tco(e.getCause().getMessage());
}
sleepSeconds(10);
pool.shutdown();
}
|
在ThreadPoolExecutor类的实现中，内部核心的任务提交方法是execute()方法，虽然用户程序通过submit()也可以提交任务，但是实际上submit()方法中最终调用的还是execute()方法。
. Executors的4种快捷创建线程池的方法
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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
| package com.crazymakercircle.mutithread.basic.create3;
import com.crazymakercircle.util.Print;
import com.crazymakercircle.util.RandomUtil;
import org.junit.Test;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import static com.crazymakercircle.util.ThreadUtil.sleepMilliSeconds;
import static com.crazymakercircle.util.ThreadUtil.sleepSeconds;
public class CreateThreadPoolDemo {
public static final int SLEEP_GAP = 500;
public static final int MAX_TURN = 5;
public static class TargetTask implements Runnable {
static AtomicInteger taskNo = new AtomicInteger(1);
protected String taskName;
public TargetTask() {
taskName = "task-" + taskNo.get();
taskNo.incrementAndGet();
}
public void run() {
Print.tco("任务：" + taskName + " doing");
sleepMilliSeconds(SLEEP_GAP);
Print.tco(taskName + " 运行结束.");
}
@Override
public String toString() {
return "TargetTask{" + taskName + '}';
}
}
static class TargetTaskWithError extends TargetTask {
public void run() {
super.run();
throw new RuntimeException("Error from " + taskName);
}
}
@Test
public void testSingleThreadExecutor() {
ExecutorService pool = Executors.newSingleThreadExecutor();
for (int i = 0; i < 5; i++) {
pool.execute(new TargetTask());
pool.submit(new TargetTask());
}
sleepSeconds(1000);
pool.shutdown();
}
@Test
public void testNewFixedThreadPool() {
ExecutorService pool = Executors.newFixedThreadPool(3);
for (int i = 0; i < 5; i++) {
pool.execute(new TargetTask());
pool.submit(new TargetTask());
}
sleepSeconds(1000);
pool.shutdown();
}
@Test
public void testNewCacheThreadPool() {
ExecutorService pool = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
pool.execute(new TargetTask());
pool.submit(new TargetTask());
}
sleepSeconds(1000);
pool.shutdown();
}
@Test
public void testNewScheduledThreadPool() {
ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(2);
for (int i = 0; i < 2; i++) {
scheduled.scheduleAtFixedRate(new TargetTask(),
0, 500, TimeUnit.MILLISECONDS);
}
sleepSeconds(1000);
scheduled.shutdown();
}
@Test
public void testNewScheduledThreadPool2() {
ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(0);
for (int i = 0; i < 2; i++) {
scheduled.scheduleAtFixedRate(new TargetTask(),
0, 500, TimeUnit.MILLISECONDS);
}
sleepSeconds(1000);
scheduled.shutdown();
}
@org.junit.Test
public void testThreadPoolExecutor() {
ThreadPoolExecutor executor = new ThreadPoolExecutor(
1,
100,
100,
TimeUnit.SECONDS,
new LinkedBlockingDeque<>(100));
for (int i = 0; i < 5; i++) {
final int taskIndex = i;
executor.execute(() ->
{
Print.tco("taskIndex = " + taskIndex);
try {
Thread.sleep(Long.MAX_VALUE);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
while (true) {
Print.tco("- activeCount:" + executor.getActiveCount() +
" - taskCount:" + executor.getTaskCount());
sleepSeconds(1);
}
}
static public class SimpleThreadFactory implements ThreadFactory {
static AtomicInteger threadNo = new AtomicInteger(1);
@Override
public Thread newThread(Runnable target) {
String threadName = "simpleThread-" + threadNo.get();
Print.tco("创建一条线程，名称为：" + threadName);
threadNo.incrementAndGet();
Thread thread = new Thread(target, threadName);
thread.setDaemon(true);
return thread;
}
}
@org.junit.Test
public void testThreadFactory() {
ExecutorService pool =
Executors.newFixedThreadPool(2, new SimpleThreadFactory());
for (int i = 0; i < 5; i++) {
pool.submit(new TargetTask());
}
sleepSeconds(10);
Print.tco("关闭线程池");
pool.shutdown();
}
public static class CustomIgnorePolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
Print.tco(r + " rejected; " + " - getTaskCount: " + e.getTaskCount());
}
}
@org.junit.Test
public void testCustomIgnorePolicy() {
int corePoolSize = 2;
int maximumPoolSize = 4;
long keepAliveTime = 10;
TimeUnit unit = TimeUnit.SECONDS;
BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(2);
ThreadFactory threadFactory = new SimpleThreadFactory();
RejectedExecutionHandler policy = new CustomIgnorePolicy();
ThreadPoolExecutor pool = new ThreadPoolExecutor(
corePoolSize,
maximumPoolSize,
keepAliveTime, unit,
workQueue,
threadFactory,
policy);
pool.prestartAllCoreThreads();
for (int i = 1; i <= 10; i++) {
pool.execute(new TargetTask());
}
sleepSeconds(10);
Print.tco("关闭线程池");
pool.shutdown();
}
private static final ThreadLocal<Long> START_TIME = new ThreadLocal<>();
@org.junit.Test
public void testHooks() {
ExecutorService pool = new ThreadPoolExecutor(2,
4, 60,
TimeUnit.SECONDS, new LinkedBlockingQueue<>(2)) {
@Override
protected void terminated() {
Print.tco("调度器已经终止!");
}
@Override
protected void beforeExecute(Thread t, Runnable target) {
Print.tco(target + "前钩子被执行");
START_TIME.set(System.currentTimeMillis());
super.beforeExecute(t, target);
}
@Override
protected void afterExecute(Runnable target, Throwable t) {
super.afterExecute(target, t);
long time = (System.currentTimeMillis() - START_TIME.get());
Print.tco(target + " 后钩子被执行, 任务执行时长（ms）：" + time);
START_TIME.remove();
}
};
pool.execute(new TargetTask());
sleepSeconds(10);
Print.tco("关闭线程池");
pool.shutdown();
}
@org.junit.Test
public void testNewFixedThreadPool2() {
ExecutorService fixedExecutorService = Executors.newFixedThreadPool(1);
ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) fixedExecutorService;
Print.tco(threadPoolExecutor.getMaximumPoolSize());
threadPoolExecutor.setCorePoolSize(8);
ExecutorService singleExecutorService = Executors.newSingleThreadExecutor();
((ThreadPoolExecutor) singleExecutorService).setCorePoolSize(8);
}
@Test
public void testShutdownGracefully() {
ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(2);
threadPool.shutdown();
try {
if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
threadPool.shutdownNow();
if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
System.err.println("线程池任务未正常执行结束");
}
}
} catch (InterruptedException ie) {
threadPool.shutdownNow();
}
}
}
|