一、引言
1.思考
先来考虑下，假如没有线程池，每次使用多线程都要创建并销毁，可能存在什么问题？
- 资源消耗：线程的创建和销毁是一项昂贵的操作，会消耗大量的 CPU 和内存资源。频繁地创建和销毁线程可能会导致资源的浪费和系统性能的下降。
- 稳定性：在创建和销毁线程的过程中，线程状态的切换可能会导致程序不稳定。例如，在线程销毁之前，可能存在一段时间的竞态条件，导致程序出现不一致的行为。
- 并发性能：在高并发场景下，频繁地创建和销毁线程会导致线程数的不稳定，进而影响程序的并发性能。过多的线程数会导致 CPU 和内存的浪费，而过少的线程数则会导致任务执行的延迟和并发性能的下降。
因此，使用线程池可以避免频繁地创建和销毁线程，从而节省资源、提高程序的稳定性和并发性能。
2.设计
既然想要利用池化技术解决线程频繁创建和销毁的问题，那么该如何设计呢？
如何实现线程的复用？
能实现线程复用的唯一方式，就是让线程不结束
那么如何让线程能够执行新的任务呢？
共享内存—>List.add()
那么线程无任务时，就一直空转吗？
有任务来时，执行任务；无任务时，阻塞
结论：
使用阻塞队列的方式，来实现线程池技术，从而完成线程的复用
3.执行流程
具体设计流程参考下图
二、源码
线程池中的核心线程是延迟初始化的
- 先初始化核心线程。
- 调用阻塞队列的方法，把task存进去。(offer() -> true/false)
- 如果true ，说明当前的请求量不大， 核心线程就可以搞定。
- false，增加工作线程(非核心线程)
- 如果添加失败，说明当前的工作线程数量达到了最大的线程数，直接调用拒绝策略。
1.构造器
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| 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.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
|
线程池创建时，这七大参数是必不可少的，但是像线程工厂和拒绝策略可不填，有默认值可用，下面详解这七个参数
corePoolSize：核心线程数，线程池中常态化保持的线程数量
maximumPoolSize：最大线程数，当核心线程与阻塞队列都满了，再有新的任务进来，而还没达到最大线程数，则会创建新的线程处理任务
keepAliveTime：超时销毁时间，超过核心线程数量的线程，休息这个时间后还没有任务处理，就会自动关闭；
unit：上一个参数的时间单位
workQueue：阻塞队列，用于保存任务，详情可参考阻塞队列
threadFactory：线程工程，用于生产线程，可采用默认值
handler：拒绝策略
- AbortPolicy（默认策略）：当任务添加到线程池中被拒绝时，将抛出RejectedExecutionException异常。
- CallerRunsPolicy：当任务添加到线程池中被拒绝时，将该任务添加到当前线程执行。
- DiscardPolicy：当任务添加到线程池中被拒绝时，将直接丢弃该任务，不做任何处理。
- DiscardOldestPolicy：当任务添加到线程池中被拒绝时，将丢弃最早加入线程池的任务，并尝试重新提交当前任务。
除了上述四种策略之外，也可以通过实现RejectedExecutionHandler接口，自定义拒绝策略。
2.execute
- 如果当前运行的线程数少于corePoolSize，尝试使用给定的命令作为新线程的第一个任务来启动一个新线程。addWorker方法会原子地检查runState和workerCount，因此可以防止出现错误警报，以避免在不需要添加线程时添加线程并返回false
- 如果可以成功将任务加入到队列中，那么我们仍然需要双重检查是否应该添加一个线程（因为现有线程在上次检查后已经死亡），或者在进入此方法后池已关闭。因此，我们重新检查状态，如果需要，在停止时回滚排队操作，或者在没有线程时启动一个新线程
- 如果无法将任务加入队列，则尝试添加一个新线程。如果失败，我们知道线程池已关闭或已饱和，因此拒绝该任务
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
| 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);
}
|
3.addWorker
尝试向线程池中添加一个新的工作线程
- 如果线程池中当前的工作线程数小于corePoolSize，则会尝试创建一个新的工作线程，并将给定的任务作为该线程的第一个任务。这个操作是原子的，即只有一个线程能够进行，避免了竞争条件。
- 如果任务不能立即分配给线程执行，会将任务放入工作队列中。
- 如果工作队列已满，将尝试创建新的工作线程来处理任务。
- 如果无法创建新的工作线程，则会根据设置的拒绝策略来处理该任务。默认的拒绝策略是抛出RejectedExecutionException异常。
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
| private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
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();
if (runStateOf(c) != rs)
continue retry;
}
}
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())
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;
}
|
这里能看到，新创建的线程进入了workers，这个容器是何方神圣呢？
1
| private final HashSet<Worker> workers = new HashSet<Worker>();
|
定义如上，用到了一个内部类Worker，如下
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
final Thread thread;
Runnable firstTask;
volatile long completedTasks;
......
......
}
|
可见，这是一个对工作线程进行包装的类，那么自然，workers中存储的也就都是创建好的线程了
4.run
在上面的Work类中，存在run方法，即线程要执行的任务
1
2
3
| public void run() {
runWorker(this);
}
|
实际上执行任务的方法，具体实现步骤如下
- 获取当前
Worker 线程对象所持有的 Thread 对象，这里的 Thread 对象在 Worker 对象的构造函数中被创建。
- 循环执行以下步骤：
- 如果
Worker 对象的 firstTask 不为 null，说明这个线程还没有开始执行任务，那么就执行 firstTask，并将 firstTask 置为 null。
- 如果
Worker 对象的 firstTask 为 null，那么就从任务队列中取出一个任务进行执行。这里的任务队列就是 ThreadPoolExecutor 的 BlockingQueue 对象，它用于存储还未被执行的任务。如果队列为空，则调用 getTask() 方法阻塞等待队列中有新的任务。
- 如果
getTask() 方法返回 null，说明任务队列已经被关闭，那么就直接跳出循环。
- 执行任务。执行任务的过程中，可能会抛出异常，这时需要调用
afterExecute() 方法进行一些清理工作，比如将异常打印出来或者记录日志。
- 如果循环结束，说明线程要么被要求退出，要么发生了异常，需要将
Worker 线程对象从线程池中移除，同时也要执行一些清理工作，比如中断当前线程、调用 processWorkerExit() 方法等。
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
| 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 ((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.getTask
用于获取任务队列中的下一个任务
两个参数：firstTask和core，分别表示线程第一次获取任务时的任务以及线程是否为核心线程
核心逻辑如下
- 如果线程池的状态为STOP，即线程池已关闭并且没有任务等待执行，那么返回null。这个判断是为了确保线程在关闭时能够正确退出。
- 如果线程为非核心线程或者线程池中正在运行的线程数大于核心线程数，那么调用workQueue的poll方法从任务队列中获取任务。如果获取到的任务不为null，则直接返回该任务。
- 如果线程为核心线程且线程池中正在运行的线程数小于等于核心线程数，则调用workQueue的take方法从任务队列中获取任务。take方法会在任务队列为空时自动阻塞线程，直到队列中有任务或线程池被关闭。如果获取到的任务不为null，则直接返回该任务。
- 如果获取到的任务为null，则返回null。
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
| private Runnable getTask() {
boolean timedOut = false;
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
|
三、线程池参数设置
- IO密集型 CPU 2core+1
- CPU利用率不高
- CPU密集型 CPU +1
- CPU利用率很高，会增加上下文切换.
