20. 线程池里的任务是怎么排队和执行的
发布于 • 阅读量 0
上一节已经把 ThreadPoolExecutor 跑起来了。
我提交了 10 个任务，但线程池里只有 3 个线程。运行时可以看到，同一时间只有 3 个任务在执行，剩下的任务并没有丢，而是在队列里等。
这一节我重点整理一个问题：
任务提交到线程池以后，到底是怎么排队、怎么被线程取走执行的？
这个问题弄清楚以后，后面再看核心线程数、最大线程数、队列容量、拒绝策略，就不会只是背参数了。
先写一个观察队列的例子
新建类：
com.succos.threadpool.ThreadPoolQueueDemo
代码如下：
package com.succos.threadpool;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolQueueDemo {
public static void main(String[] args) {
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2,
2,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(10),
new ThreadPoolExecutor.CallerRunsPolicy()
);
for (int i = 1; i <= 6; i++) {
int taskId = i;
executor.execute(() -> {
System.out.println(Thread.currentThread().getName()
+ " 开始执行任务 "
+ taskId);
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
System.out.println(Thread.currentThread().getName()
+ " 执行完成任务 "
+ taskId);
});
System.out.println("提交任务 " + taskId
+ " 后，队列中等待任务数："
+ executor.getQueue().size());
}
executor.shutdown();
try {
executor.awaitTermination(1, TimeUnit.HOURS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("全部任务执行完成");
}
}
这里我把线程池设置成：
new ThreadPoolExecutor(
2,
2,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(10),
new ThreadPoolExecutor.CallerRunsPolicy()
);
意思是：
核心线程数：2；
最大线程数：2；
队列容量：10。
所以这个线程池最多同时有 2 个线程在执行任务。
我提交了 6 个任务，那么大概会出现这种情况：
任务1、任务2 被线程执行；
任务3、任务4、任务5、任务6 进入队列等待。
运行时大概会看到什么
控制台输出可能类似这样：
提交任务 1 后，队列中等待任务数：0
提交任务 2 后，队列中等待任务数：0
提交任务 3 后，队列中等待任务数：1
提交任务 4 后，队列中等待任务数：2
提交任务 5 后，队列中等待任务数：3
提交任务 6 后，队列中等待任务数：4
pool-1-thread-1 开始执行任务 1
pool-1-thread-2 开始执行任务 2
等待 3 秒...
pool-1-thread-1 执行完成任务 1
pool-1-thread-1 开始执行任务 3
pool-1-thread-2 执行完成任务 2
pool-1-thread-2 开始执行任务 4
具体顺序不一定完全一样，但现象是差不多的。
重点是：
任务不是全部同时执行；
前两个任务被线程执行；
后面的任务先进入队列；
线程执行完当前任务后，再从队列里取下一个任务。
任务提交后，线程池先看什么？
先看这个配置：
corePoolSize = 2
maximumPoolSize = 2
workQueue = new ArrayBlockingQueue<>(10)
我提交第 1 个任务时，线程池发现当前工作线程数还没到核心线程数 2。
于是它会创建一个线程来执行任务 1。
提交第 2 个任务时，当前线程数是 1，还没到核心线程数 2。
所以继续创建第二个线程来执行任务 2。
提交第 3 个任务时，当前线程数已经等于核心线程数 2。
这时两个核心线程都在忙。
于是任务 3 不会马上执行，而是进入任务队列。
后面的任务 4、5、6 也是一样，都会进入队列等待。
所以这个例子的执行规则大概是：
核心线程没满：创建线程执行；
核心线程满了：任务进队列等。
任务在队列里等的是什么？
这个地方我以前也容易想错。
我一开始会觉得，是不是线程在排队？
其实不是。
在这个场景里，排队的是任务，也就是一个个 Runnable。
比如我提交了：
executor.execute(() -> {
System.out.println("处理任务");
});
这里的 Lambda 本质上就是一个 Runnable 任务。
当线程池暂时没空执行它时，这个 Runnable 会被放进 workQueue。
也就是这里：
new ArrayBlockingQueue<>(10)
所以队列里放的不是线程，而是任务。
线程池里的工作线程执行完当前任务后，会去队列里取下一个任务继续执行。
Worker 线程怎么工作
可以用一个简化模型理解线程池里的工作线程。
线程池里的线程大概会做这种事：
while (true) {
Runnable task = workQueue.take();
task.run();
}
当然，真实源码比这个复杂很多，但学习阶段这样理解就够了。
它的意思是：
线程执行完一个任务后，不会马上销毁；
它会继续去队列里取下一个任务；
如果队列里没有任务，它就在队列那里等；
等新任务来了，再取出来执行。
这就是线程复用。
如果每个任务都 new Thread()，任务执行完线程就结束了。
线程池不是这样。
线程池里的线程是长期干活的工人，任务来了就干，干完再取下一个。
为什么说线程池有“任务队列”
手动 new Thread() 时，我没有任务队列这个概念。
我写：
new Thread(task).start();
就是马上创建一个线程去执行这个任务。
而线程池是：
executor.execute(task);
这行代码不是简单地“马上创建线程”。
它会根据当前线程池状态决定：
直接创建线程执行；
或者放到队列里；
或者队列满了以后再扩容；
或者最后执行拒绝策略。
所以线程池不是简单替代 new Thread()。
它多了一个调度过程。
这个调度过程的核心就是：
线程数量有限；
任务可以排队；
线程空闲后再取任务。
用 PDF 处理来理解队列
假设我有 10 个 PDF：
a.pdf
b.pdf
c.pdf
d.pdf
e.pdf
f.pdf
g.pdf
h.pdf
i.pdf
j.pdf
线程池配置是 3 个线程。
那么执行时更像这样：
pdf-thread-1 处理 a.pdf
pdf-thread-2 处理 b.pdf
pdf-thread-3 处理 c.pdf
d.pdf、e.pdf、f.pdf、g.pdf、h.pdf、i.pdf、j.pdf 在队列里等
等 pdf-thread-1 处理完 a.pdf：
pdf-thread-1 从队列里取 d.pdf 继续处理
等 pdf-thread-2 处理完 b.pdf：
pdf-thread-2 从队列里取 e.pdf 继续处理
所以线程池里真正稳定存在的是这几个工作线程。
PDF 任务只是不断排队、被取走、执行完成。
这个模型比每个 PDF 都创建一个线程要稳得多。
getQueue().size() 只是观察用
代码里我用了：
executor.getQueue().size()
这个可以看到当前队列里大概有多少个任务在等待。
但这个值在多线程环境下是实时变化的。
比如我刚打印时队列里有 4 个任务，下一瞬间某个线程执行完当前任务，马上从队列里取走一个，队列数量就变成 3。
所以这个值适合学习和观察，不适合在业务逻辑里依赖它做精确判断。
我这里用它只是为了看清楚任务提交后有没有进入队列。
队列满了会怎么样？
现在队列容量是 10，任务只有 6 个，所以不会满。
为了观察队列满了以后的情况，可以把队列容量改小一点。
比如：
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2,
2,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2),
new ThreadPoolExecutor.AbortPolicy()
);
这里线程池最多 2 个线程，队列最多放 2 个任务。
如果我提交 6 个任务，大概是：
任务1、任务2：被两个线程执行；
任务3、任务4：进入队列；
任务5：线程满了，队列也满了，触发拒绝策略；
任务6：同样触发拒绝策略。
如果拒绝策略是 AbortPolicy，就会直接抛异常：
java.util.concurrent.RejectedExecutionException
这就说明，线程池不是无限接收任务。
线程数量和队列容量都有限。
这个限制在真实项目里非常重要。
一个触发拒绝的例子
可以新建一个类：
com.succos.threadpool.ThreadPoolQueueRejectDemo
代码如下：
package com.succos.threadpool;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolQueueRejectDemo {
public static void main(String[] args) {
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2,
2,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2),
new ThreadPoolExecutor.AbortPolicy()
);
for (int i = 1; i <= 6; i++) {
int taskId = i;
System.out.println("准备提交任务 " + taskId);
executor.execute(() -> {
System.out.println(Thread.currentThread().getName()
+ " 开始执行任务 "
+ taskId);
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
System.out.println(Thread.currentThread().getName()
+ " 执行完成任务 "
+ taskId);
});
System.out.println("提交任务 " + taskId
+ " 后，队列中等待任务数："
+ executor.getQueue().size());
}
executor.shutdown();
}
}
这段代码大概率会在提交第 5 个任务时抛异常。
因为：
两个线程正在执行任务1、任务2；
队列里已经放了任务3、任务4；
任务5 再来时，线程池已经没有地方放它了。
这时就触发拒绝策略。
为什么 maximumPoolSize 在这里没发挥作用？
上面的例子里，我设置的是：
corePoolSize = 2
maximumPoolSize = 2
核心线程数和最大线程数一样。
所以线程池最多只能有 2 个线程。
当核心线程满了以后，任务先进入队列。
队列满了以后，因为最大线程数也已经到 2 了，不能再创建新线程，所以直接拒绝。
如果我把最大线程数改成 4：
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2,
4,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2),
new ThreadPoolExecutor.AbortPolicy()
);
那执行规则就会变成：
任务1、任务2：创建核心线程执行；
任务3、任务4：进入队列；
任务5、任务6：队列满了，但线程数还没到 maximumPoolSize，于是创建非核心线程执行；
后面再来任务，如果线程也到 4 了，队列也满了，才会拒绝。
这个规则下一节还会继续展开。
这里先记住一点：
线程池不是一开始就创建到最大线程数；
它通常是先用核心线程，再用队列，队列满了才考虑最大线程数。
这个顺序很容易记错。
线程池的任务处理顺序
我现在把线程池提交任务后的处理顺序简单记成这样：
1. 当前线程数小于 corePoolSize：创建核心线程执行任务；
2. 当前线程数达到 corePoolSize：任务进入队列；
3. 队列满了，并且当前线程数小于 maximumPoolSize：创建非核心线程执行任务；
4. 队列满了，线程数也达到 maximumPoolSize：执行拒绝策略。
这四步非常重要。
后面所有线程池参数配置，基本都围绕这个规则展开。
特别是第 2 步和第 3 步：
不是核心线程满了就马上创建到最大线程数；
而是核心线程满了以后，先尝试进入队列。
只有队列也满了，才会创建非核心线程。
PDF 项目里怎么理解队列容量
在 PDF 批量处理里，队列容量不能随便写。
如果队列太小，任务稍微多一点就容易触发拒绝。
如果队列太大，任务会大量堆积，虽然不拒绝，但可能导致内存压力变大，用户也一直等不到结果。
比如：
new ArrayBlockingQueue<>(100)
表示最多排队 100 个 PDF 任务。
如果我的系统一次最多就处理几十个 PDF，这可能够用。
但如果可能一次提交几千个文件，就要重新设计了。
不能只是把队列改成很大，比如：
new ArrayBlockingQueue<>(100000)
这看起来不容易拒绝，但实际上是在把压力堆到内存里。
更合理的方式可能是：
任务入库；
后台线程池分批消费；
接口返回任务 ID；
前端轮询任务状态。
这就已经是任务系统设计的问题了。
但线程池队列这个点，是后面设计的基础。
这一节小结
这一节我主要记住几点：
1. 线程池里排队的是任务，不是线程；
2. execute() 提交的是 Runnable 任务；
3. 核心线程忙不过来时，任务会进入 workQueue；
4. 工作线程执行完当前任务后，会继续从队列里取下一个任务；
5. 队列满了以后，才会考虑创建非核心线程；
6. 队列满了、线程也达到最大数量时，会触发拒绝策略；
7. getQueue().size() 可以帮助学习观察，但不要在业务里依赖它做精确判断。
用一句话总结：
线程池不是让所有任务一起跑，而是用有限线程反复消费队列里的任务。
下一节继续看线程池的拒绝策略。
因为只要线程池和队列都有上限，就一定要考虑：任务真的放不下时，到底怎么办。
