并发编程六、线程池原理解析
2022/8/31 1:24:02
本文主要是介绍并发编程六、线程池原理解析，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
前言：
- 文章内容：线程与进程、线程生命周期、线程中断、线程常见问题总结
- 本文章内容来源于笔者学习笔记，内容可能与相关书籍内容重合
- 偏向于知识核心总结，非零基础学习文章，可用于知识的体系建立，核心内容复习，如有帮助，十分荣幸
- 相关文献：并发编程实战、计算机原理
为何要用线程池？
- 创建/销毁线程，是个重资源的操作，为了避免频繁的创建和销毁线程，让创建的线程进行复用，就用了线程池
- 在一些需要多线程和缓冲任务的场景下，线程池能够提供缓冲和线程管理机制
在实际项目中使用线程池的场景？
- 根据系统的承受能力，调整线程池中工作线线程的数目，防止因为消耗过多的内存，而把服务器累趴下
- 定时任务系统里用线程池：在可以预计任务执行时间的情况下，可以设置每个任务错开时间执行，保证多个线程池高峰期不重合。
- 业务用线程池：根据内存大小和任务的大小，合理的给一个阻塞队列值，然后根据业务类型来走拒绝策略，回到正常的业务系统
线程池的七大核心参数：
- 核心线程数：默认情况下一直存在于线程池中，即使没有任务执行
- 线程池最大线程数：非核心线程+核心线程数总和。如果队列已经满了，则会创建新的线程，如果线程数超过了该值，会执行拒绝策略(如果是无界队列，该参数就失效了)。注意：在ThreadPoolExecutor内部，并不区分哪个线程是核心，哪个是非核心，超过了核心线程数量的线程会被销毁，那么剩下的就是核心线程了。
- 非核心线程闲置超时时长：非核心线程处于闲置状态时间超过该值，就被销毁。设置allowCoreThreadTimeOut(true)也可作用于核心线程
- 超时时长单位：微毫秒(NANOSECONDS)、微秒(MICROSECONDS)、毫秒（MILLSECONDS）、秒（MINUTES）
- 阻塞队列：存放等待执行的线程任务对象
- 线程工厂：创建线程的工厂，批量创建线程。统一在创建线程时设置一些参数。如果不指定，会新建一个默认的线程工厂
- 拒绝处理策略：线程数量大于最大线程数就会采用拒绝处理策略
-
- AbortPolicy：默认策略，丢弃任务并抛出异常
- DiscardPolicy：丢弃新任务，但不抛异常
- DiscardOlderstPolicy：丢弃队列中最旧的任务，在执行execute方法
- CallerRunsPolicy：直接用当前业务线程来执行任务
线程池状态：
- RUNNING：线程池创建后处于RUNNING状态
- SHUTDOWN：调用shutdown方法进入该状态，线程池不接受新任务，清除空闲工作线程，等待阻塞队列中的任务完成
- STOP：调用shutdownNow方法进入该状态，线程池不接受新任务，中断所有线程，阻塞队列中未执行的任务全部丢弃
- TIDYING：所有任务已终止，ctl记录的任务数量为0，进入该状态，然后执行terminated函数
- TERMINATED：执行完terminated函数后进入该状态
线程池execute执行过程：
四种常用线程池：
- newCachedThreadPool：核心池大小为0，最大线程数是Integer.MAX_VALUE。任务创建就进入SynchronousQueue中。超时设置60s，适用于执行短时间任务，线程复用率较高，不会占用很多资源。
- newFixedThreadPool：核心池与最大线程数大小一致，创建的都是核心线程，采用LinkedBlockingQueue大小Integer.MAX_VALUE。由于创建核心线程，没有任务时在getTask方法中会阻塞在take方法上，线程不会被回收，这时占用资源也更多。几乎不会触发拒绝策略，阻塞队列很大。CachedThreadPool因为最大线程数超大，也几乎不会触发拒绝策略。
- newSingleThreadExecutor：有且仅有一个核心线程，使用LinkedBlockingQueue，不会创建非核心线程，任务按照先来先执行顺序，唯一线程不空闲，任务在队列中等待。
- newScheduledThreadPool：定长线程池，支持定时及周期性任务执行。
阻塞队列：
- ArrayBlockingQueue：数组结构有界队列，先进先出原则，新元素插入到队列尾部，获取从队列头开始获取元素
- LinkedBlockingQueue：链表构成无界队列，默认大小Integer.MAX_VALUE，先进先出
- DelayQueue：只有当元素指定延迟时间到了，才能从队列中获取元素，无大小限制
- PriorityBlockingQueue：基于优先级的无界阻塞队列，生产不阻塞，消费阻塞。使用时生产数据速度必能快于消费速度，时间长会耗尽堆空间。放入元素必须实现Comparable接口，排序值高的永远排在前面
- SynchronousQueue：没有内部容量，put一个元素必须等待take取数据，然后生产者会阻塞
手写阻塞队列：
private List<Integer> container = new ArrayList<>();
private volatile int size;
private volatile int capacity;
private Lock lock = new ReentrantLock();
//condition
private final Condition notEmpty = lock.newCondition();
private final Condition notFull = lock.newCondition();
MyBlockQueue(int cap){
this.capacity = cap;
}
public void put(int data){
try{
lock.lock();
try {
while (size > capacity){
System.out.println("队列满了");
notEmpty.await();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
++size;
container.add(data);
notFull.signal();
}finally {
lock.unlock();
}
}
public int take(){
try{
lock.lock();
try {
while (size==0){
System.out.println("队列空了");
notFull.await();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
--size;
Integer ret = container.get(0);
container.remove(0);
notEmpty.signal();
return ret;
}finally {
lock.unlock();
}
}
如何针对不同类型的任务来使用线程池呢？
- 高并发、任务执行时间短：任务耗时短，要求线程尽量少，如果线程太多，有可能出现线程切换和管理的时间，大于任务执行的时间，那效率就低了
- 并发不高，任务执行时间长：如果业务时间长集中在IO操作，那就是IO密集型任务，IO操作不占用CPU，所以不要让所有CPU闲下来，可以加大线程池中线程数目，让CPU处理更多业务。如果业务时间长集中在计算上，那就是计算密集型任务，线程池中线程数设置少一点，减少线程上下文切换
- 并发高，任务执行时间长：考虑具体任务类型，是IO密集还是CPU密集，线程池这块要控制好线程数和队列容量，不然线程池很容易满。改善任务，对任务执行时间进行优化，考虑某些数据是否可以走缓存，不重要的业务是否可以用中间件拆分解耦，异步处理+回调降低执行时间。
源码分析
带着问题看源码：
如何在线程执行前后额外操作、线程池如何复用线程，线程池如何保证线程不会销毁，超过核心线程数的线程是如何销毁，能够实现创建的线程数达到最大线程数再将任务放入队列吗核心属性ctl与线程池状态
//int类型，前三位标识线程池状态，后29位标识有效线程的数量 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); //29位，方便ctl进行位运算使用的常量 private static final int COUNT_BITS = Integer.SIZE - 3; //线程池的线程最大容量 private static final int CAPACITY = (1 << COUNT_BITS) - 1; //线程池运行状态 前三位 private static final int RUNNING = -1 << COUNT_BITS;//运行中：111 private static final int SHUTDOWN = 0 << COUNT_BITS;//线程被shutdown，继续执行完剩下的任务 000 private static final int STOP = 1 << COUNT_BITS;//线程被shutdownNow，线程池停止，中断所有任务 001 private static final int TIDYING = 2 << COUNT_BITS;//shutdown或shutdownNow后，任务都被处理完，到这个过渡状态 010 private static final int TERMINATED = 3 << COUNT_BITS;//线程池停止 011
execute：执行方法
public void execute(Runnable command) {
//非空判断
if (command == null)
throw new NullPointerException();
//获取ctl变量
int c = ctl.get();
//workerCountOf()：获取线程池中正在工作的线程数 判断是否小于核心数
if (workerCountOf(c) < corePoolSize) {
//核心线程还有，创建核心线程，并执行任务。传入true代表是核心线程
if (addWorker(command, true))
return;
//添加核心线程数失败，重新获取ctl
c = ctl.get();
}
//如果线程池运行状态是运行中，将任务追加到阻塞队列
if (isRunning(c) && workQu eue.offer(command)) {
//添加阻塞队列成功
int recheck = ctl.get();
//二次校验线程池状态 如果不是运行中，并移除成功，就执行拒绝策略
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
//如果工作线程为0，创建一个工作线程
addWorker(null, false);
}
//没有核心线程且阻塞队列满了 添加有一个非核心线程 创建成功返回true，失败返回false
else if (!addWorker(command, false))
//添加最大线程数失败，执行拒绝策略
reject(command);
}
addWorker()：创建工作线程
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {//外循环判断线程池状态
//获取ctl 和runStateOf()：获取当前线程池运行状态
int c = ctl.get();
int rs = runStateOf(c);
//rs >= SHUTDOWN：说明线程池执行了shutdown或shutdownNow
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN && //线程池停了
firstTask == null && //任务为null
! workQueue.isEmpty())) //队列空了
return false;//返回false，不创建工作线程
for (;;) {//内存循环判断线程池工作线程数量
//获取当前线程池中正在工作的线程数
int wc = workerCountOf(c);
//工作线程大于核心线程 或 工作线程大于最大线程 返回false，不创建工作线程
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//满足可以创建工作线程的条件，CAS操作：正在工作的线程数+1。这里没有真正的创建工作线程
if (compareAndIncrementWorkerCount(c))
break retry;
//重新获取ctl
c = ctl.get();
//CAS失败，就要判断是重新执行内循环，还是外循环。如果线程池状态改变，则执行外循环。否则执行内循环
if (runStateOf(c) != rs)
continue retry;
}
}
//声明两个标识 workerStarted：工作线程启动了吗 workerAdded：工作线程添加了吗
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//new一个worker，并传入任务
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
//创建工作线程时，通过ReentrantLock来保证多线程并发创建，获取了线程池的全局锁
final ReentrantLock mainLock = this.mainLock;
//上锁，shutdown、shutdownNow也需要获取全局锁
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
//线程池状态是否是运行状态
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {//或者线程池为shutdown并且任务是null
if (t.isAlive()) //线程正在运行，抛出异常。这里工作线程还没有创建和启动，不会是运行。
throw new IllegalThreadStateException();
//将工作线程追加到workers，worker是一个HaseSet<Worker>。
workers.add(w);
int s = workers.size();
//如果现在工作线程数大于历史最大值，替换掉历史最大值
if (s > largestPoolSize)
largestPoolSize = s;
//修改工作线程添加标记为true，标识工作线程创建好了
workerAdded = true;
}
} finally {
//释放锁
mainLock.unlock();
}
//如果工作线程添加成功率
if (workerAdded) {
//启动线程
t.start();
//修改工作线程启动标识为true
workerStarted = true;
}
}
} finally {
//如果工作线程启动失败
if (! workerStarted)
//补救操作
addWorkerFailed(w);
}
return workerStarted;//返回工作线程的启动结果
}
Work对象：封装工作线程
Worker继承了AQS，实现了Runnable。private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
//存放线程对象
final Thread thread;
//存放任务
Runnable firstTask;
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); //设置state 在runWorker之前禁止中断worker创建过程。
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
//调用worker的start方法，实际上执行的worker中的run方法，即runWorker方法
public void run() {
runWorker(this);
}
runWorker方法：
- 实现线程执行前后额外增强：beforeExecute和afterExecute
- 线程复用：runWorker方法中，有个while循环，会一直判断任务不为null或队列中是否有任务。有新任务就调用task.run执行。不会再去创建新线程，来达到复用的目的。
final void runWorker(Worker w) {
//获取当前线程
Thread wt = Thread.currentThread();
//获取worker中具体任务
Runnable task = w.firstTask;
//将worker中任务设置为null
w.firstTask = null;
//这里unlock实际调用时aqs的release(1)，释放构造方法设置的标识。代表当前线程可以被打断
w.unlock();
//设置标记为true 最后都为true表示：线程执行或获取任务过程发生异常
boolean completedAbruptly = true;
try {
//如果worker中task有任务，直接执行。如果没有任务，就从队列中取任务getTask()
while (task != null || (task = getTask()) != null) {
//要执行任务，添加一个标记。即便shutdown也不会打断任务执行
w.lock();
//判断当前线程池状态以及线程状态，是否需要被中断。这时候线程池都不能执行任务了，就会执行中断
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//执行任务之前执行beforeExecute：前置增强
beforeExecute(wt, task);
Throwable thrown = null;
try {//这个的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：后置增强，这样可以让开发者可以在线程执行之前和之后添加额外处理。
afterExecute(task, thrown);
}
} finally {
//线程执行完了，任务设置为null，任务处理完毕0
task = null;
//当前worker处理的任务数+1
w.completedTasks++;
//去掉标记
w.unlock();
}
}
//没有任务进来了，队列任务也执行完了，就是false
completedAbruptly = false;
} finally {
//当任务没有了或队列中也没有任务了执行processWorkerExit。销毁线程
processWorkerExit(w, completedAbruptly);
}
}
getTask：
在getTsak时，会判断工作线程数是否大于最大线程数、核心线程是否设置了可销毁、队列是否为空。来判断是否要CAS的减少工作线程数。如果线程池是非运行状态并且工作队列是空的会CAS的减少工作线程数，并销毁线程。private Runnable getTask() {
//设置超时标记
boolean timedOut = false;
for (;;) {
//获取运行状态
int c = ctl.get();
int rs = runStateOf(c);
//如果线程池非运行状态 并且 （线程池已经停止或工作队列是空的）
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
//CAS的减少工作线程数
decrementWorkerCount();
return null;//返回null
}
//获取工作中的线程数
int wc = workerCountOf(c);
//allowCoreThreadTimeOut：false核心线程空闲也存活，ture核心线程使用keepAliveTime超时等待工作
//工作线程数大于核心线程数或核心线程可销毁，timed就是true
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//如果工作线程大于最大线程数 或者 工作线程数大于最大线程数了，或核心线程可销毁
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
//工作线程数至少大于1，是因为timed为true代表工作队列可能还有任务，所以至少要保留一条线程执行任务
// 或者 工作队列空了那可以直接销毁 cas减少工作线程数，成功返回null
if (compareAndDecrementWorkerCount(c))
return null;
continue;//CAS失败，再次循环
}
try {
//如果timed为true 则超时的去队列获取任务，否则阻塞获取
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
//如果返回不是null,说明队列有任务
return r;
timedOut = true;//超时标记为true，表示超时时间到了，队列中还没有任务。继续执行循环
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
processWorkerExit：
- 线程销毁时：会校验线程池状态试图终止线程池，然后看是否是正常退出，如果异常退出新增一个工作线程继续处理。
- 如果是正常退出，代表队列没任务且任务执行完了，然后判断是否设置了核心线程可销毁，设置了最小线程数为0，没有就核心线程数。
- 如果最小线程数为0了，但队列不为空，就保留一个线程继续执行任务
private void processWorkerExit(Worker w, boolean completedAbruptly) {
//为true 表示线程池在执行任务或取队列任务执行过程发生了异常
if (completedAbruptly)
//CAS的把工作线程数减为0
decrementWorkerCount();
//获取全局锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//线程池完成的任务数进行叠加
completedTaskCount += w.completedTasks;
//将线程移除出worker的set
workers.remove(w);
} finally {
//释放全局锁
mainLock.unlock();
}
//校验线程池的状态，试图终止线程池
tryTerminate();
int c = ctl.get();
//线程池状态小于stop 即运行或shutdown状态
if (runStateLessThan(c, STOP)) {
//线程池是正常退出，即队列没有任务，且任务执行完了
if (!completedAbruptly) {
//如果设置了核心线程可销毁 则最新线程为0，否则是核心线程数
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
//如果min是0 并且队列不为空
if (min == 0 && ! workQueue.isEmpty())
//保留一个线程执行任务
min = 1;
if (workerCountOf(c) >= min)
return;//如果工作线程数大于等于最小线程数，返回
}
//如果是异常退出走这里，创建一个线程
//如果当前线程数量<最小最小线程数，也要创建一个线程
addWorker(null, false);
}
}
线程池如何实现线程复用：
线程池对Thread做了包装，不重复调用Thread.start。而是自己有一个Runnable.run，run方法在循环里跑。跑得过程中不断检查是否有新加入的子Runnable对象，有的话就调用自己的run。相当于把一个大run把其他小run串联起来。同一个Thread可以执行不同的runnable，线程池把线程和Runnable通过队列解耦了，线程从队列中不断获取新任务线程池的线程是如何不保证不被销毁的：
线程池的线程都被封装成了Worker，在Worker内部会循环的从队列获取任务，如果获取的任务为空，则销毁线程，如果任务不为空，则执行任务。从队列获取任务时，如果当前线程数量没有超过核心线程数量，会阻塞式的获取任务，超过了或者设置了核心线程可超过会超时时的获取任务，超时时间为KeepAliveTime，超时时间获取不到就会返回空个，则会结束线程。即通过一个循环保证线程不会结束运行，从而保证不被销毁。线程池是如何动态调整参数的：
- ThreadPoolExecutor.setCorePoolSize // 修改核心线程数
- ThreadPoolExecutor.setMaximumPoolSize // 修改最大线程数
- ThreadPoolExecutor.setKeepAliveTime // 修改空闲线程存活时间
- ThreadPoolExecutor.setRejectedExecutionHandler // 修改拒绝策略
- ThreadPoolExecutor.setThreadFactory // 修改线程工厂
超过核心线程数的线程是如何销毁的：
如果超过了核心线程数，在获取任务时会超时式的从队列获取任务，如果早keepAliveTime时间后没有任务返回，会返回null，会导致while循环退出，从而走到了销毁线程的逻辑。能够实现创建的线程数达到最大线程数再将任务放入队列吗：
在execute方法第二步，如果workQueue.offer返回了false，就会创建新线程来执行任务。只需要重写队列的offer方法。实现如果当前线程数量没有达到最大线程数量时，返回false，走创建线程逻辑，而不是将任务放入队列。
这篇关于并发编程六、线程池原理解析的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
- 2024-12-27数据结构与算法面试题详解及练习
- 2024-12-27网络请求面试题详解与实战
- 2024-12-27数据结构和算法面试真题详解与实战教程
- 2024-12-27网络请求面试真题解析与实战教程
- 2024-12-27数据结构和算法大厂面试真题详解与实战指南
- 2024-12-27TS大厂面试真题解析与应对策略
- 2024-12-27TS大厂面试真题详解与解析
- 2024-12-27网站安全入门：如何识别和修复漏洞
- 2024-12-27SQL注入基础教程
- 2024-12-27初学者指南：理解和修复跨域漏洞
