🏷️个人主页:牵着猫散步的鼠鼠
🏷️系列专栏:Java全栈-专栏
🏷️本系列源码仓库:多线程并发编程学习的多个代码片段(github)
🏷️个人学习笔记,若有缺误,欢迎评论区指正
目录
1.前言
1.1.内容回顾
1.2.本节任务
2.实现思路
2.1 线程工厂实现思路
2.2 核心线程与非核心线程实现思路
3.代码实现
3.1.线程池工厂实现
3.2核心线程与非核心线程逻辑
4.总结
✨️本系列源码均已上传仓库 1321928757/Concurrent-MulThread-Demo(github.com)✨️
(本章节可参考liushijie-240329-core分支)
1.前言
1.1.内容回顾
往期文章传送门:
Java多线程实战-从零手搓一个简易线程池(一)定义任务等待队列-CSDN博客Java多线程实战-从零手搓一个简易线程池(二)线程池与拒绝策略实现-CSDN博客
在上一节我们实现了线程池内部的基本运转逻辑,池化了线程资源进行任务处理,细心的同学可以发现,我们上章没有划分核心线程与非核心线程的概念,在JDK官方的提供的线程池中,线程池中的线程从概念上分为核心线程和非核心线程,核心线程是线程池中长久存在的线程,默认不会被回收,而非核心线程在空闲时间超过设置的最大空闲时间时会被回收,当然,我们也可以通过设置一个属性来运行核心线程被回收。
1.2.本节任务
本章节的任务如下:
- 实现线程工厂
- 实现核心线程与非核心线程
2.实现思路
2.1 线程工厂实现思路
线程工厂是运用了工厂设计模式,可以帮助我们隐藏创建线程的一些细节。我们可以通过线程工厂在创建线程数时定义线程的一些属性,如线程名称、线程组等。实现线程工厂一般有以下步骤:
- 定义一个线程工厂接口或抽象类,提供创建新线程的方法。
- 实现该接口或继承该抽象类,重写创建线程的方法逻辑。
- 在线程池的构造函数中,传入自定义的线程工厂实例。
整体实现还是比较简单,主要就是要注意编码规范
2.2 核心线程与非核心线程实现思路
这里首先要清楚一个概念,JDK线程池源码中没有显式的区别核心线程和非核心线程,他只是线程池在处理线程池不同情况下的线程的一种概念。我们接下来从源码分析(JDK1.8)是如何实现核心线程和非核心线程的管理的。
JDK官方线程池中的runWorker方法作用是用来执行worker线程
final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {while (task != null || (task = getTask()) != null) {// 线程执行任务流程,省流}completedAbruptly = false;} finally {processWorkerExit(w, completedAbruptly);}}
同我们上节运行线程一样,他会通过while (task != null || (task = getTask()) != null)来重复获取任务,如果task == null,也就是没获取到,会进入到processWorkerExit函数中,线程会被回收。也就是说,只要getTask方法返回为null,就代表了当前线程需要回收,所以我们接下来重点查看getTask方法的源码:
private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?// 1.方法内部使用了一个无限循环for (;;),这意味着线程会一直尝试获取任务,直到成功获取到任务或者满足退出条件。for (;;) {// 2.获取到目前线程池的线程数,最大核心线程,最大总线程数等信息int c = ctl.get();int rs = runStateOf(c);// 3.如果线程池的运行状态至少为SHUTDOWN(在此状态以上的状态,都不会接受新任务了,所以我们直接返回null)if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();return null;}int wc = workerCountOf(c);//获取线程池当前线程数量// 4.根据当前线程数动态判断是否要回收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;}}}
getTask方法主要负责从workQueue队列中获取任务,如果获取到了就返回任务,如果没有获取到就返回null。他会根据线程池的当前状态,当前线程数,来动态的选择是否从workQueue中拿取任务,以及拿取操作是否是超时操作。这里的设计特别巧妙,建议阅读源码仔细体会
如果 当前线程数 > 最大核心线程数,我们就判定存在非核心线程,可以进行回收判断
如果 当前线程数 < 最大线程数,我们就判定不存在核心线程
所以核心线程和非核心线程他们都是一类线程,只是在线程池不同情况下划分的概念恶意
3.代码实现
3.1.线程池工厂实现
3.1.1.线程工厂接口
/*** @author Luckysj @刘仕杰* @description 线程工厂接口* @create 2024/03/28 20:40:18*/
public interface ThreadFactory {/*** @description* @param * @return 创建的线程对象* @date 2024/03/28 21:01:35*/Thread newThread(Runnable r);
}
3.1.2.默认线程工厂实现类
默认线程工厂实现类主要是设置新建线程的线程组,线程名前缀等等信息,更加规范,方便后续日志排查错误
/*** @author Luckysj @刘仕杰* @description 默认线程工厂,我们这里仿照源码写法,为每个线程分配线程组(默认会自动分配),并为每个线程组* @create 2024/03/28 21:27:10*/
public class DefaultThreadFactory implements ThreadFactory{/** 原子序号类,我们可以通过该类为线程工厂来获取一个随机序号,主要是为了区分不同线程池实例*/private static final AtomicInteger poolNumber = new AtomicInteger(1);/** 线程组,每个线程都需要属于一个线程组(平常使用未指定线程组会默认分配)*/private final ThreadGroup group;/** 原子序号类,我们可以通过该类为每个线程来获取一个随机序号*/private static final AtomicInteger threadNumber = new AtomicInteger(1);/** 线程名前缀,以便于在日志、监控等场景下识别和管理线程。*/private final String namePrefix;public DefaultThreadFactory() {// 获取管理安全策略的类,通过这个类我们可以获取对应名称的线程组,SecurityManager 和 group 的存在是为了更好地控制线程的安全性和权限SecurityManager s = System.getSecurityManager();// 存在 SecurityManager实例,则通过 s.getThreadGroup() 获取一个受限制的线程组。// 如果不存在 SecurityManager 实例,则使用当前线程所在的线程组 Thread.currentThread().getThreadGroup()。this.group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();// 生成前缀this.namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";}@Overridepublic Thread newThread(Runnable r) {Thread thread = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);// 将线程设置为用户线程if(thread.isDaemon()){thread.setDaemon(false);}// 为线程设置默认优先级if(thread.getPriority() != Thread.NORM_PRIORITY){thread.setPriority(Thread.NORM_PRIORITY);}return thread;}
}
3.1.3.使用线程工厂
在Worker工作线程构造函数中使用工厂创建线程
class Worker implements Runnable{private Runnable firstTask;private Thread thread;public Worker(Runnable task) {this.firstTask = task;this.thread = threadFactory.newThread(this);}// 省略}
3.2核心线程与非核心线程逻辑
3.2.1.编写getTask方法
getTask方法会根据线程池情况动态从任务队列中获取任务
/*** @description 从等待队列中获取任务* @return Runnable 待执行的任务,没有获取到会返回null* @date 2024/04/02 10:46:37*/
public Runnable getTask(){//我们使用一个变量来记录上次循环获取任务是否超时boolean preIsTimeOut = false;// 内部使用一个while循环,线程会一直尝试获取任务,直到成功获取到任务或者满足退出条件while(true){// 获取线程池当前线程数量int wc = threadTotalNums.get();// 1.是否要进行核心线程回收操作,当allowCoreThreadTimeOut为true,或者当前线程池数大于核心线程数时,我们需要进行回收判断boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;// 2.根据情况动态调整线程数,以下情况需要直接返回null(返回null就会回收线程):// (1)当前线程大于最大线程数(就是超过规定大小了),且任务队列为空且存在工作线程// (2)timed为true,上次任务超时了(preIsTimeOut = true),且任务队列为空且存在工作if ( (wc > maximumPoolSize || (timed && preIsTimeOut)) && (wc > 1 || workQueue.isEmpty()) ) {return null;}// 3.根据timed这个条件来选择是超时堵塞Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;// 获取任务超时了,将preIsTimeOut设为true,下次可以执行回收preIsTimeOut = true;}}
- timed 变量决定了线程从等待队列中拿取任务的方式,如果当前线程数大于最大核心线程数,或者开启了允许核心线程回收(allowCoreThreadTimeOut = true),我们就超时拿取,这样如果拿取任务超时就会返回null,线程就会被回收
3.2.2.调整Worker工作线程的run方法
将原来直接从任务队列中获取任务改为通过getTask方法获取
@Overridepublic void run() {log.info("工作线程====》工作线程{}开始运行", Thread.currentThread());// 1。首先消费当前任务,消费完再去任务队列取,while循环实现线程复用while(firstTask != null || (firstTask = getTask()) != null){try {firstTask.run();}catch (Exception e){throw new RuntimeException(e);}finally {// 执行完后清除任务firstTask = null;}}// 2.跳出循环,说明取任务超过了最大等待时间,线程歇菜休息吧synchronized (workerSet){workerSet.remove(this);threadTotalNums.decrementAndGet(); //计数扣减}log.info("工作线程====》线程{}已被回收,当前线程数:{}", Thread.currentThread(), threadTotalNums.get());}
3.2.3.编写addWorker方法
/*** @description 添加工作线程* @param firstTask 线程第一次执行的任务* @param isCore 是否为核心线程* @return Boolean 线程是否添加成功* @date 2024/04/02 10:42:43*/public Boolean addWorker(Runnable firstTask, Boolean isCore){if(firstTask == null) {throw new NullPointerException();}// TODO 1.我们在添加线程时,首先可以进行一些与线程池生命周期相关的校验,比如在一些状态下,不允许再添加任务// 2.根据当前线程池和isCore条件判断是否需要创建int wc = threadTotalNums.get();if (wc >= (isCore ? corePoolSize : maximumPoolSize))return false;// 3.创建线程,并添加到线程集合中Worker worker = new Worker(firstTask);Thread t = worker.thread;if(t != null){synchronized (workerSet){workerSet.add(worker);threadTotalNums.getAndIncrement();}t.start();return true;}return false;}
3.2.4.完善excute方法
流程如下:
1.如果当前线程数小于核心线程,直接创建核心线程去运行
2.线程数大于核心线程,我们就将任务加入等待队列
3.队列满了,尝试创建非核心线程,如果失败就触发拒绝策略
public void execute(Runnable task){if(task == null){throw new NullPointerException("传递的Runnable任务为Null");}// 1.如果当前线程数小于核心线程,直接创建线程去运行if(threadTotalNums.get() < corePoolSize){if(addWorker(task, true)) return;}// 2.线程数大于核心线程,我们就将任务加入等待队列if(workQueue.offer(task)){return;}// 3.队列满了,尝试创建非核心线程,如果失败就触发拒绝策略else if(!addWorker(task, false)){reject(task);}}
4.测试
编写如下测试代码,我们会创建一个核心线程数为2,最大线程数为5,等待队列长度为5的线程池,并添加15个任务到线程池中,按照预期会有五个任务触发拒绝策略,在任务执行完成后只保留两个核心线程
@Slf4j
public class MainTest {public static void main(String[] args) {ThreadPool threadPool = new ThreadPool(new WorkQueue<>(5), 2, 5,5L, TimeUnit.SECONDS,(queue, task) -> {log.info("拒绝策略====》拒绝策略触发,直接丢弃当前任务");}, new DefaultThreadFactory());threadPool.setAllowCoreThreadTimeOut(false); //不回收核心线程for (int i = 0; i < 15; i++) {threadPool.execute(() -> {System.out.println("执行任务------->当前执行线程为" + Thread.currentThread().toString());try {Thread.sleep(5000);} catch (InterruptedException e) {throw new RuntimeException(e);}});}// ExecutorService executorService = Executors.newFixedThreadPool(2);}
}
运行结果如下:
可以看到运行结果符合预期,任务也被正常消费
我们设置AllowCoreThreadTimeOut的属性为true,再次进行测试,
threadPool.setAllowCoreThreadTimeOut(true); //回收核心线程
结果输出:
可以看到,核心线程也会被回收,符合预期。
5.总结
在本章节中我们通过学习JDK线程池源码中的部分代码,实现了一个简易版带有核心线程与非核心线程处理逻辑的线程池,我们可以通过指定AllowCoreThreadTimeOut属性来设置是否允许核心线程的回收,默认只会回收非核心线程。线程池的官方源码还是写得相当巧妙的,阅读难度也不高,推荐小伙伴学习~