关于作者：CSDN内容合伙人、技术专家， 从零开始做日活千万级APP。
专注于分享各领域原创系列文章 ，擅长java后端、移动开发、人工智能等，希望大家多多支持。
目录
- 一、导读
- 二、概览
- 2.1 为什么创建和销毁线程开销较大
- 2.2 为什么要使用线程池？
- 2.3 在配置线程池的时候需要考虑哪些配置因素?
- 三、使用
- 3.1 线程池的创建
- 3.1.1 newFixedThreadPool
- 3.1.2 newCachedThreadPool
- 3.1.3 newScheduledThreadPool
- 3.1.4 newSingleThreadExecutor
- 3.2 ThreadPoolExecutor
- 3.3 为什么线程池不推荐使用Executors去创建，而是通过ThreadPoolExecutor的方式
- 四、原理
- 4.1 线程池中常用的workQueue（缓冲队列）
- 4.2 线程池中拒绝策略
- 4.3 任务的关闭
- 4.4 线程的复用
- 五、 推荐阅读
一、导读
我们继续总结学习Java基础知识，温故知新。
二、概览
在Java中，创建和销毁线程开销较大，为了避免线程过多而带来使用上的开销。
所以我们需要对线程进行统一管理及复用，这就是我们要说的线程池。
线程池用于管理和复用多个线程，把一个或多个线程通过统一的方式进行调度和重复使用的技术。
从JDK 5开始，把工作单元与执行机制分离开来，工作单元包括Runnable和Callable，而执行机制由Executor框架提供。
2.1 为什么创建和销毁线程开销较大
创建和销毁线程的开销较大主要是因为涉及到以下几个方面：
-
上下文切换：在多线程环境中，当一个线程被创建或销毁时，操作系统需要切换上下文，将CPU的执行权从一个线程转移到另一个线程。这个过程涉及保存和恢复线程的状态信息，包括寄存器值、栈指针和程序计数器等。上下文切换是一项耗时的操作，会导致额外的开销。
-
内存管理：每个线程需要分配一定的内存空间来存储线程的堆栈、线程私有数据等。创建和销毁线程会涉及内存的分配和释放，而内存分配和释放操作通常比较耗时。
-
调度开销：操作系统需要进行调度，决定哪个线程应该获得CPU的执行权。线程的创建和销毁会引起调度器的重新调度，这涉及到时间片、优先级和调度算法等方面的开销。
-
同步和通信：多线程编程中，线程之间通常需要进行同步和通信，以确保数据的一致性和线程间的协调。创建和销毁线程会涉及到锁、信号量、管道等同步和通信机制的初始化和清理，增加了开销。
2.2 为什么要使用线程池？
- 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
- 提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
- 提高线程的可管理性。
2.3 在配置线程池的时候需要考虑哪些配置因素?
从任务的优先级，任务的执行时间长短，任务的性质(CPU密集/ IO密集)，任务的依赖关系这四个角度来分析。并且近可能地使用有界的工作队列。
性质不同的任务可用使用不同规模的线程池分开处理:
- CPU密集型: 尽可能少的线程，Ncpu+1
- IO密集型: 尽可能多的线程, Ncpu*2，比如数据库连接池
- 混合型: CPU密集型的任务与IO密集型任务的执行时间差别较小，拆分为两个线程池；否则没有必要拆分
CPU密集型和IO密集型任务的权衡：如何找到最佳平衡点
三、使用
3.1 线程池的创建
java提供了多种方式：
1. newFixedThreadPool(n):创建一个数量固定的线程池，超出的任务会在队列中等待空闲的线程，
可用于控制程序的最大并发数。
2. newCacheThreadPool():短时间内处理大量工作的线程池，会根据任务数量产生对应的线程，
并试图缓存线程以便重复使用，如果限制 60 秒没被使用，则会被移除缓存。如果现有线程没有可用的，
则创建一个新线程并添加到池中，如果有被使用完但是还没销毁的线程，就复用该线程。
终止并从缓存中移除那些已有 60 秒钟未被使用的线程。因此，长时间保持空闲的线程池不会使用任何资源。
3. newScheduledThreadPool()：创建一个数量固定的线程池，支持执行定时性或周期性任务。
4. newWorkStealingPool(n)：Java 8 新增创建线程池的方法，创建时如果不设置任何参数，
则以当前机器CPU 处理器数作为线程个数，此线程池会并行处理任务，不能保证执行顺序。
5. newSingleThreadExecutor()：创建一个单线程的线程池。这个线程池只有一个线程在工作，
也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。
此线程池保证所有任务的执行顺序按照任务的提交顺序执行。
6. newSingleThreadScheduledExecutor()：此线程池就是单线程的 newScheduledThreadPool。
3.1.1 newFixedThreadPool
创建固定大小的线程池，比如线程池容量是10，最多可以同时执行10个线程。
使用案例
创建线程池,参数是创造的线程数量
ExecutorService pool = Executors.newFixedThreadPool(10);
for (int i = 0; i < 10; i++) {
int j = i;
pool.execute(new Runnable() {
@Override
public void run() {
}
});
}
3.1.2 newCachedThreadPool
创建一个可缓存的线程池，此线程池不会对线程池大小做限制，线程池大小完全依赖于JVM能够创建的最大线程大小，当然线程池里的线程是可以复用的，但是如果在高并发的情况下，这个线程池在会导致运行时内存溢出问题。
使用案例
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 6; i++) {
int j = i;
executorService.execute(new Runnable() {
@Override
public void run() {
}
});
}
3.1.3 newScheduledThreadPool
创建一个定时执行的线程池，里边提供了两个方法，FixRate和fixDelay，
fixRate 就是以固定时间周期执行任务，不管上一个线程是否执行完，
fixDelay 的话就是以固定的延迟执行任务，就是在上一个任务执行完成之后，延迟一定时间执行。
使用案例
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(2);
for (int i = 0; i < 5; i++) {
int j = i;
executorService.schedule(new Runnable() {
@Override
public void run() {
}
}, 3L, TimeUnit.SECONDS);
}
3.1.4 newSingleThreadExecutor
创建一个单线程的线程池，这个线程池同时只能执行一个线程，可以保证线程按顺序执行，保证数据安全。
使用案例
public class SingleThreadPoolDemo {
//格式化
static SimpleDateFormat sim = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//AtomicInteger用来计数
static AtomicInteger number = new AtomicInteger();
public static void main(String[] args) {
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i = 0; i < 6; i++) {
executorService.execute(new Runnable() {
@Override
public void run() {
}
});
}
}
}
3.2 ThreadPoolExecutor
通过ThreadPoolExecutor的方式创建线程池，前面四种都不是我们推荐都方式
public class ThreadPoolDemo {
private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
2, // 核心线程数
10, // 最大线程数
10L, // 线程存活时间
TimeUnit.SECONDS, // 线程存活时间单位
new LinkedBlockingQueue(100));// 缓冲队列
public static void main(String[] args) {
threadPoolExecutor.execute(new Runnable() {
@Override
public void run() {
}
});
}
}
3.3 为什么线程池不推荐使用Executors去创建，而是通过ThreadPoolExecutor的方式
newFixedThreadPool（固定线程数）
newSingleThreadExecutor（单线程）
- 主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM。线程数固定，任务多了之后容易堆积。
newCachedThreadPool（可缓存的线程池）
newScheduledThreadPool（定时执行的线程池）
- 主要问题是线程数最大数是Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至OOM。不限定线程多数量，任务一多，容易创建无限多线程。
四、原理
先看个图，方便理解
- 当有新的任务进来时，线程池将当前线程数量与核心数量进行比较，如果没有超过核心数就会新建线程进行任务执行，
- 如果已经超过核心线程数，则判断缓冲队列是否已经满了，没有满的话任务就会被放入缓冲队列中排队等待执行；
- 如果缓冲队列超过最大队列数，并且线程池没有达到最大线程数，就会新建线程来执行任务；
- 如果超过了最大线程数，就会执行拒绝执行策略。
再简单点，就两个队列，一个线程集合workerSet和一个阻塞队列workQueue
我们一起来看看源码，
线程池类为 java.util.concurrent.ThreadPoolExecutor，常用构造方法为：
ThreadPoolExecutor(
int corePoolSize, // 核心线程数
int maximumPoolSize, // 最大线程数
long keepAliveTime, // 线程存活时间
TimeUnit unit, // 线程存活时间单位
BlockingQueue<Runnable> workQueue, // 缓冲队列
RejectedExecutionHandler handler // 拒绝策略
)
处理任务的优先级为：核心线程 > 缓冲队列 > 最大线程
corePoolSize > workQueue > maximumPoolSize
如果三者都满了，使用handler处理被拒绝的任务。
/**
* 将该Runnable任务加入线程池并在未来某个时刻执行
* 该任务可能执行在一个新的线程 或 一个已存在的线程池中的线程
* 如果该任务提交失败,可能是因为线程池已关闭,或者已达到线程池队列和线程数已满.
* 该Runnable将交给RejectedExecutionHandler处理,抛出RejectedExecutionException
*/
public void execute(Runnable command) {
if (command == null){
//如果没传入Runnable任务,则抛出空指针异常
throw new NullPointerException();
}
int c = ctl.get();
//当前线程数 小于 核心线程数
if (workerCountOf(c) < corePoolSize) {
//直接开启新的线程,并将Runnable传入作为第一个要执行的任务,成功返回true,否则返回false
if (addWorker(command, true)){
return;
}
c = ctl.get();
}
//c < SHUTDOWN代表线程池处于RUNNING状态 + 将Runnable添加到任务队列,如果添加成功返回true失败返回false
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//成功加入队列后,再次检查是否需要添加新线程(因为已存在的线程可能在上次检查后销毁了,或者线程池在进入本方法后关闭了)
if (! isRunning(recheck) && remove(command)){
//如果线程池处于非RUNNING状态 并且 将该Runnable从任务队列中移除成功,则拒绝执行此任务
//交给RejectedExecutionHandler调用rejectedExecution方法,拒绝执行此任务
reject(command);
}else if (workerCountOf(recheck) == 0){
//如果线程池线程数量为0,则创建一条新线程,去执行
addWorker(null, false);
}
}else if (!addWorker(command, false))
//如果线程池处于非RUNNING状态 或 将Runnable添加到队列失败(队列已满导致),则执行默认的拒绝策略
reject(command);
}
当ThreadPoolExecutor创建新线程时，通过CAS来更新线程池的状态ctl.
4.1 线程池中常用的workQueue（缓冲队列）
-
ArrayBlockingQueue(有界缓存等待队列)
可以指定缓存队列的大小 -
LinkedBlockingQueue(无界缓存等待队列)
可以创建Integer.MAX_VALUE个线程，容易OOM
当前执行的线程数量达到corePoolSize（核心）的数量时，剩余的元素会在阻塞队列里等待，在使用此阻塞队列时maximumPoolSizes就相当于无效了。
- SynchronousQueue(无缓冲等待队列)
是一个不存储元素的阻塞队列，会直接将任务交给消费者，必须等队列中的添加元素被消费后才能继续添加新的元素。使用SynchronousQueue阻塞队列一般要求maximumPoolSizes为无界，避免线程拒绝执行操作
4.2 线程池中拒绝策略
所有拒绝策略都实现了接口 RejectedExecutionHandler
public interface RejectedExecutionHandler {
/**
* @param r 待执行任务
* @param executor 线程池
* @throws RejectedExecutionException 方法可能会抛出拒绝异常
*/
void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}
1.AbortPolicy
直接抛出拒绝异常，会中断调用者的处理过程，所以除非有明确需求，一般不推荐
public static class AbortPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
2.CallerRunsPolicy
在调用者线程中运行当前被丢弃的任务,也就是说谁把 runnable 这个任务甩出来。
用调用者所在线程来运行任务，也就是说任务不会进入线程池。
如果线程池已经被关闭，则直接丢弃该任务。
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
- DiscardOledestPolicy
丢弃队列中最老的，然后再次尝试提交新任务。
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
//获得待执行的任务队列,队列先进先出
//poll()方法就能直接把队列中最老的抛弃掉，再次尝试执行execute(r)
e.getQueue().poll();
e.execute(r);
}
}
}
- DiscardPolicy
默默丢弃无法加载的任务。这个代码就很简单了，真的是啥也没做。
public static class DiscardPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
- 自定义拒绝策略
只要继承接口都可以根据自己需要自定义拒绝策略.
案例1：
单独启动一个新的临时线程来执行任务。
private class NewThreadRunsPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
try {
final Thread t = new Thread(r, "Temporary task executor");
t.start();
} catch (Throwable e) {
throw new RejectedExecutionException(
"Failed to start a new thread", e);
}
}
}
案例2：
直接继承的 AbortPolicy ，加强了日志输出，并且输出dump文件,然后任务也是拒绝
public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
String msg = String.format("Thread pool is EXHAUSTED!" +
" Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d)," +
" Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",
threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),
e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
url.getProtocol(), url.getIp(), url.getPort());
logger.warn(msg);
dumpJStack();
throw new RejectedExecutionException(msg);
}
}
4.3 任务的关闭
shutdown方法会将线程池的状态设置为SHUTDOWN,线程池进入这个状态后,就拒绝再接受任务,然后会将剩余的任务全部执行完
- 临时线程什么时候创建？
新任务提交时发现核心线程都在忙，任务队列也满了，并且还可以创建临时线程，
此时才会创建临时线程； - 什么时候会开始拒绝任务？
核心线程和临时线程都在忙，任务队列也满了，新的任务过来的时候才会开始拒绝任务
4.4 线程的复用
在ThreadPoolExecutor.java的runwork方法中通过一个while循环,不断的getTask()取任务出来执行,以这种方式实现了线程的复用.
五、 推荐阅读
Java 专栏
SQL 专栏
数据结构与算法
Android学习专栏
