Java多线程实战-从零手搓一个简易线程池(三)线程工厂,核心线程与非核心线程逻辑实现

🏷️个人主页:牵着猫散步的鼠鼠 

🏷️系列专栏: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.本节任务

本章节的任务如下:

  1. 实现线程工厂
  2. 实现核心线程与非核心线程

2.实现思路

2.1 线程工厂实现思路

线程工厂是运用了工厂设计模式,可以帮助我们隐藏创建线程的一些细节。我们可以通过线程工厂在创建线程数时定义线程的一些属性,如线程名称、线程组等。实现线程工厂一般有以下步骤:

  1. 定义一个线程工厂接口或抽象类,提供创建新线程的方法。
  2. 实现该接口或继承该抽象类,重写创建线程的方法逻辑。
  3. 在线程池的构造函数中,传入自定义的线程工厂实例。

整体实现还是比较简单,主要就是要注意编码规范

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属性来设置是否允许核心线程的回收,默认只会回收非核心线程。线程池的官方源码还是写得相当巧妙的,阅读难度也不高,推荐小伙伴学习~

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/587868.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

flink on yarn

前言 Apache Flink&#xff0c;作为大数据处理领域的璀璨明星&#xff0c;以其独特的流处理和批处理一体化模型&#xff0c;成为众多企业和开发者的首选。它不仅能够在处理无界数据流时展现出卓越的实时性能&#xff0c;还能在有界数据批处理上达到高效稳定的效果。本文将简要…

网络协议学习——HTTPS

目录 ​编辑 一&#xff0c;认识HTTPS 二&#xff0c;加密方式 1&#xff0c;对称式加密 2&#xff0c;非对称式的加密 3&#xff0c;数据指纹&#xff08;数据摘要&#xff09; 4&#xff0c;数据签名 三&#xff0c;HTTPS的工作原理 实现方式 数字证书 一&#xff0c…

pygame--坦克大战(二)

加载敌方坦克 敌方坦克的方向是随机的&#xff0c;使用随机数生成。 初始化敌方坦克。 class EnemyTank(Tank):def __init__(self,left,top,speed):self.images {U: pygame.image.load(img/enemy1U.gif),D: pygame.image.load(img/enemy1D.gif),L: pygame.image.load(img/e…

Golang学习系列1-pprof性能调优

1. pprof 简述 一位亦师亦友的话让我记忆犹新&#xff0c;他说“学习一个新事务&#xff0c;应该从三个方面入手what,why,how;且三者的重要程度应该是递减”。所以在本文的第一部分先叙述下pprof的what & why。 1.1 What&#xff1f; pprof是golang自身提供的一种性能分…

大模型量化技术-GPTQ

大模型量化技术-GPTQ 2022年,Frantar等人发表了论文 GPTQ:Accurate Post-Training Quantization for Generative Pre-trained Transformers。 这篇论文详细介绍了一种训练后量化算法,适用于所有通用的预训练 Transformer模型,同时只有微小的性能下降。 GPTQ算法需要通过…

剑指offer打卡 JZ6 从尾到头打印链表

在牛客网刷的&#xff0c;还是跟leetcode一样非acm模式&#xff0c;由于急着暑期实习题量不固定&#xff0c;八股算法轮刷 打卡内容偏个人笔记&#xff0c;本人水平一般(代码随想录稀里糊涂刷了一遍)&#xff0c;从小白开始分析(甚至会分析语法)&#xff0c;尽量一题多解深入探…

SpringBoot | Spring Boot“整合Redis“

目录: 1. Redis 介绍2. Redis 下载安装3. Redis “服务开启”和“连接配置”4. Spring Boot整合Redis的“前期准备” :① 编写实体类② 编写Repository 接口③ 在“全局配置文件”中添加 “Redis数据库” 的 “相关配置信息” 5. Spring Boot整合“Redis” (案例展示) 作者简介…

不同设备使用同一个Git账号

想要在公司和家里的电脑上用同一个git账号来pull, push代码 1. 查看原设备的用户名和邮箱 第1种方法&#xff0c; 依次输入 git config user.name git config user.email第2种方法&#xff0c; 输入 cat ~/.gitconfig2. 配置新设备的用户名和邮箱 用户名和邮箱与原设备保持…

C-偶遇行军蚁(遇到过的题,做个笔记)

我的代码: 思路就是把每一行看成一个字符串&#xff0c;然后逐渐增加字符就行 #include <iostream> #include <vector> using namespace std; int main() {string s;int n;cin >> n; //读入行数cin >> s; //读入字符串vector<string>arr(n…

黄金票据制作

1、黄金票据 黄金票据在内网渗透中主要用于权限维持&#xff0c;即留下后门。使用黄金票据前提是你已经控制了整个域的域控制器。其核心原理与服务器中的krbtgt用户相关。从以下可以了解krbtgt原理&#xff1a; 前言 | windows protocolhttps://daiker.gitbook.io/windows-pro…

开源布道 共创生态——2024 AutoMQ 布道师计划启动!

我们是来自 Apache RocketMQ 和 Linux LVS 项目的核心团队&#xff0c;曾经见证并应对过消息队列基础设施在大型互联网公司和云计算公司的挑战。基于对象存储优先、存算分离、多云原生等技术理念&#xff0c;我们重新设计并实现 Apache Kafka 和 Apache RocketMQ&#xff0c;带…

Vue2(完结):replace属性、编程式路由导航、缓存路由组件、两个新钩子、路由守卫、history与hash

一、router-link的replace属性 1、作用&#xff1a;控制路由跳转时操作浏览器历史记录的模式 2、浏览器的历史记录有两种写入方式&#xff1a;分别为push和replace&#xff0c;push是追加历史记录&#xff0c;replace是替换当前记录。路由跳转时候默认为push 3、如何开启repla…