Java 自定义线程池实现

自定义线程池

  • 简介
  • 任务图示
  • 阻塞队列 BlockingQueue<T>
    • ReentrantLock
    • 代码
  • 线程池 ThreadPool
    • 工作线程类 Worker
  • 拒绝策略接口
  • 代码测试类 TestThreadPool
    • 为什么需要j = i?(lambad表达式相关)
  • 测试结果
    • 拒绝策略:让调用者自己执行任务
    • 拒绝策略:让调用者放弃任务执行

简介

学习 黑马JUC 并发编程 过程中根据教程完成的线程池案例

任务图示

在这里插入图片描述
线程池Thread Pool保存工作资源,即n个工作线程,通过从任务队列Blocking Queue中获取m个任务执行。

阻塞队列 BlockingQueue

注意ReentrantLock类的使用

包括任务队列、生产者消费者锁、容量capcity、阻塞获取、带时阻塞获取、阻塞添加、带时阻塞添加、带拒绝策略的添加

Blocking Queue对获取任务或者添加任务的操作进行阻塞管理,分别设置生产者锁(任务队列满则阻塞生产者,停止添加任务)、消费者锁(任务队列空则阻塞消费者,停止获取任务)。

ReentrantLock

基本语法:

reentrantLock.lock();
try{// 临界区
} finally {// 释放锁reentrantLock.unlock();
}

创建条件变量,以及 阻塞 和 唤醒 的基本使用
在这里插入图片描述

代码

@Slf4j(topic = "c.BlockingQueue")
class BlockingQueue<T> {// 1. 任务队列private Deque<T> queue = new ArrayDeque<>();// 2. 锁private ReentrantLock lock = new ReentrantLock();// 3. 生产者条件变量private Condition fullWaitSet = lock.newCondition();// 4. 消费者条件变量private Condition emptyWaitSet = lock.newCondition();// 5. 容量private int capcity;public BlockingQueue(int capcity) {this.capcity = capcity;}// 带超时阻塞获取public T poll(long timeout, TimeUnit unit) {lock.lock();try {//将timeout统一转换为 纳秒long nanos = unit.toNanos(timeout);//如果任务队列为空,在消费者阻塞队列中,超时时间内循环阻塞while(queue.isEmpty()){try{if (nanos <= 0){return null;}//超时时间内进行 空队列阻塞nanos = emptyWaitSet.awaitNanos(nanos);} catch (InterruptedException e) {throw new RuntimeException(e);}}T t = queue.removeFirst();//唤醒 生产者阻塞队列 中的线程fullWaitSet.signal();return t;}finally{lock.unlock();}}// 阻塞获取public T take() {lock.lock();try{//循环阻塞while(queue.isEmpty()){try {emptyWaitSet.await();}catch (InterruptedException e) {throw new RuntimeException(e);}}T t = queue.removeFirst();fullWaitSet.signal();return t;}finally{lock.unlock();}}// 阻塞添加public void put(T task) {lock.lock();try{while(queue.size() >= capcity){try{log.debug("等待加入任务队列————任务:{} ", task);fullWaitSet.await();} catch (InterruptedException e) {throw new RuntimeException(e);}}log.debug("加入任务队列————任务: {}", task);queue.addLast(task);emptyWaitSet.signal();}finally{lock.unlock();}}// 带超时时间阻塞添加public boolean offer(T task, long timeout, TimeUnit timeUnit) {lock.lock();try{long nanos = timeUnit.toNanos(timeout);while(queue.size() >= capcity){if(nanos <= 0) {return false;}try {log.debug("等待加入任务队列————任务:{} ", task);//awaitNanos返回剩余等待时间nanos = fullWaitSet.awaitNanos(nanos);} catch (InterruptedException e) {throw new RuntimeException(e);}}log.debug("加入任务队列————任务: {}", task);queue.addLast(task);emptyWaitSet.signal();return true;}finally{lock.unlock();}}public int size() {lock.lock();try {return queue.size();} finally {lock.unlock();}}public void tryPut(RejectPolicy<T> rejectPolicy, T task) {lock.lock();try{//判断队列是否满if(queue.size() >= capcity){//使用拒绝策略rejectPolicy.reject(this, task);} else{log.debug("加入任务队列————任务: {}", task);queue.addLast(task);emptyWaitSet.signal();}}finally{lock.unlock();}}
}

线程池 ThreadPool

包含 阻塞队列 、工作线程集合、线程数量、超时时间设置、执行任务

//线程池
@Slf4j(topic = "c.ThreadPool")
class ThreadPool {// 任务队列private BlockingQueue<Runnable> taskQueue;// 线程集合private HashSet<Worker> workers = new HashSet<>();// 核心线程数private int coreSize;// 获取任务时的超时时间private long timeout;private TimeUnit timeUnit;private RejectPolicy<Runnable> rejectPolicy;//初始化public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity,RejectPolicy<Runnable> rejectPolicy) {this.coreSize = coreSize;this.timeout = timeout;this.timeUnit = timeUnit;this.taskQueue = new BlockingQueue<>(queueCapcity);this.rejectPolicy = rejectPolicy;}// 执行任务public void execute(Runnable task) {synchronized (workers){if(workers.size() < coreSize){Worker worker = new Worker(task);log.debug("新增 worker 投入工作————工作线程:{}, 任务:{}", worker, task);workers.add(worker);worker.start();}else{taskQueue.tryPut(rejectPolicy, task);}}}class Worker extends Thread{...}
}

工作线程类 Worker

线程池ThreadPool的内部类,主要设置了线程的执行方法,
没有设置超时时间,线程执行完所有任务之后不会被销毁,而是继续等待更多任务,设置超时时间线程则会在长时间获取不到任务后结束执行
while(task != null || (task = taskQueue.take()) != null)循环获取任务队列中的任务并执行
while(task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) 为带超时的执行方法,循环到超时时间则停止循环

  • 一个线程与一个任务(不带超时)
    在这里插入图片描述

  • 一个线程与一个任务(带超时)
    在这里插入图片描述

class Worker extends Thread{private Runnable task;public Worker(Runnable task){this.task = task;}@Overridepublic void run() {//task不为空执行任务,执行完毕再从任务队列中获取任务并执行//while(task != null || (task = taskQueue.take()) != null){//超时设置while(task != null || (task = taskQueue.poll(timeout, timeUnit)) != null){try {log.debug("线程正在执行————任务:{}", task);task.run();} catch(Exception e){e.printStackTrace();} finally {task = null;}}synchronized (workers){log.debug("worker 被移除————工作线程:{}", this);workers.remove(this);}}}

拒绝策略接口

在任务队列满时设置不同的策略处理之后再申请添加的队列

@FunctionalInterface // 拒绝策略
interface RejectPolicy<T> {void reject(BlockingQueue<T> queue, T task);
}

代码测试类 TestThreadPool

初始化ThreadPool时使用lambda表达式实现拒绝策略,threadPool.execute(() -> { } 传入的就是任务的内容,设置Thread.sleep(1000L) 用于测试拒绝策略

@Slf4j(topic = "c.TestThreadPool")
public class TestThreadPool{public static void main(String[] args) {ThreadPool threadPool = new ThreadPool(2,1000, TimeUnit.MILLISECONDS, 2, (queue, task)->{// 1. 死等// queue.put(task);// 2) 带超时等待// queue.offer(task, 1500, TimeUnit.MILLISECONDS);// 3) 让调用者放弃任务执行// log.debug("放弃{}", task);// 4) 让调用者抛出异常// throw new RuntimeException("任务执行 失败 " + task);// 5) 让调用者自己执行任务task.run();});for (int i = 0; i < 10; i++) {int j = i;threadPool.execute(() -> {try {Thread.sleep(1000L);} catch (InterruptedException e) {e.printStackTrace();}log.debug("{}", j);});}}
}

为什么需要j = i?(lambad表达式相关)

Lambda表达式内部引用的变量必须是final或者是事实上的final(即在Lambda表达式内部没有改变过它的值)。在Java 8之前,匿名内部类内部引用的局部变量必须声明为final。在Java 8中,这一规则放宽了一些,但要求这些变量的值在编译时确定,所以被称为"effectively final"(事实上的final)。

i虽然在每次循环中都有不同的值,但它是在Lambda表达式内部引用的。由于Lambda表达式的执行是延迟的,当Lambda表达式被执行时,循环可能已经结束,i的值可能已经改变。为了避免这种情况,需要将i赋值给一个final或者事实上的final变量j,以确保Lambda表达式内部引用的变量是不可变的。

因此,通过将i赋值给j,保证了在Lambda表达式内部引用的j是不可变的,从而避免了可能出现的并发问题。

测试结果

拒绝策略:让调用者自己执行任务

跟如上测试代码一样:coreSize 工作线程数 2,queueCapctiy 任务队列容量 2 ,循环执行10个任务,采用 task.run() 主线程执行的拒绝策略

任务0、1被线程执行,2、3 放入任务队列中,之后也被线程执行,其余线程则被主线程直接执行
在这里插入图片描述

拒绝策略:让调用者放弃任务执行

任务0、1被线程执行,2、3 放入任务队列中,之后也被线程执行,其余线程则被放弃
在这里插入图片描述

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

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

相关文章

求解线性方程组

如图题意看出x1有且仅有两种可能&#xff0c;1或者0&#xff0c;且知道了所有a的值&#xff0c;且因为要求所得答案字典序最小&#xff0c;所以先假设x10。 又因a2x1x2所以可以求出x2的值&#xff0c;又如a2x1x2x3,所以可以求出x3的值依次求出所有x的值&#xff0c;但每求出一…

20.Python从入门到精通—参数 位置参数 关键字参数 默认参数 匿名函数 return 语句 强制位置参数

20.从入门到精通:参数 位置参数 关键字参数 默认参数 匿名函数 return 语句 强制位置参数 参数位置参数关键字参数默认参数 匿名函数return 语句强制位置参数 参数 在Python中&#xff0c;函数可以接受任意数量的参数&#xff0c;包括位置参数、关键字参数和默认参数。以下是这…

卷积神经网络基础

卷积层 1.可参考此文章&#xff1a;https://blog.csdn.net/tjlakewalker/article/details/83275322 2.实现代码: import torch.nn as nn conv nn.Conv2d(in_channels3, #输入通道out_channels64, #输出通道kernel_size3, #卷积核stride1) #步长 print(conv) # 结…

Optional 的用法

常见的有三种方法创建Optional 创建一个为空的optional实例 Testpublic void test_create_empty_optional() {Optional<String> empty Optional.empty();Assert.assertFalse(empty.isPresent());} 用of 创建的Option实例必须为非空&#xff0c;用ofNullable创建的Opti…

SpringCloud-记

目录 什么是SpringCloud 什么是微服务 SpringCloud的优缺点 SpringBoot和SpringCloud的区别 RPC 的实现原理 RPC是什么 eureka的自我保护机制 Ribbon feigin优点 Ribbon和Feign的区别 什么是SpringCloud Spring Cloud是一系列框架的有序集合。它利用Spring Boot的开发…

2024/3/24周报

文章目录 摘要Abstract文献阅读题目引言创新点数据处理研究区域和数据缺失值处理水质相关分析 方法和模型LSTMAttention机制AT-LSTM模型 实验结果 深度学习transformer代码实现1 模型输入1.1 Embedding层1.2 位置编码 2 Encoder2.1 编码器2.2 编码器层2.3注意力机制2.4多头注意…

hive SQL 移位、运算符、REGEXP正则等常用函数

orderflag & shiftleft(1,14) shiftleft(1,14) SQL中使用的运算符号详解_sql中各种符号-CSDN博客 Hive函数_hive shift-CSDN博客 &#xff08;内建函数&#xff08;类型排序&#xff09;_云原生大数据计算服务 MaxCompute(MaxCompute)-阿里云帮助中心&#xff09;

第五讲 哈希表

我们在前面讲了存储层&#xff0c;以及从次磁盘中将页面加载到缓冲池【Buffer Pool】中&#xff0c;现在我们继续往上&#xff0c;来讨论如何支持 DBMS 的执行引擎从页面中读取/写入数据。这部分是访问方法层的功能&#xff0c;它负责通过索引或者表本身&#xff0c;设置是其他…

为什么sql优化往往重点考虑查,而不是增删改

SQL优化重点考虑查询而不是增删改的原因有以下几点&#xff1a; 查询操作的频率高&#xff1a;在大多数数据库应用中&#xff0c;查询操作的频率远高于增删改操作。用户经常需要从数据库中获取数据进行分析、展示或进一步处理。因此&#xff0c;优化查询操作可以显著提高整个系…

Web基础应用

Web基础应用 独立的Web服务 只提供一个页面 构建基本的Web服务 基于 B/S &#xff08;Browser/Server&#xff09;架构的网页服务 服务端提供网页 浏览器下载并显示网页 Hyper Text Markup Language&#xff0c;超文标记语言 Hyper Text Transfer Protocol&#xff0c;超…

Spring JdbcTemplate 案例详解教程

&#xff08;感谢您的关注和点赞支持&#xff01;&#xff09; Spring JdbcTemplate Spring JdbcTemplate 是 Spring Framework 中用于简化 JDBC 操作的一个轻量级封装工具类。它提供了一系列的方法来执行 SQL 查询、更新、批处理等操作&#xff0c;从而减轻了开发者手动处理…

生物信息学文章中常见的图应该怎么看?

目录 火山图 热图 箱线图 森林图 LASSO回归可视化图&#xff08;套索图&#xff09; 交叉验证图 PCA图 ROC曲线图 这篇文章只介绍这些图应该怎么解读&#xff0c;具体怎么绘制&#xff0c;需要什么参数&#xff0c;怎么处理数据&#xff0c;会在下一篇文章里面给出 火山…

Python | 八、类型注解

类型注解 类型注解&#xff08;Type Annotations&#xff09;在Python 3.5及更高版本中被引入。类型注解允许你为函数的参数和返回值指定预期的数据类型。这些注解不会对Python代码的运行时行为产生直接影响&#xff0c;也就是说&#xff0c;它们不会导致代码在运行时进行类型…

python之jsonpath的使用

文章目录 介绍安装语法语法规则举例说明 在 python 中使用获取所有结构所有子节点的作者获取所有子孙节点获取所有价格取出第三本书的所有信息取出价格大于70块的所有书本从mongodb 中取数据的示例 介绍 JSONPath能在复杂的JSON数据中 查找和提取所需的信息&#xff0c;它是一…

Java设计模式之单例模式(多种实现方式)

虽然写了很多年代码&#xff0c;但是说真的对设计模式不是很熟练&#xff0c;虽然平时也会用到一些&#xff0c;但是都没有深入研究过&#xff0c;所以趁现在有空练下手 这章主要讲单例模式&#xff0c;也是最简单的一种模式&#xff0c;但是因为spring中bean的广泛应用&#…

前端面试题《react》

说说React render方法的原理&#xff1f;在什么时候会被触发&#xff1f; render函数里面可以编写JSX&#xff0c;转化成createElement这种形式&#xff0c;用于生成虚拟DOM&#xff0c;最终转化成真实DOM 在 React 中&#xff0c;类组件只要执行了 setState 方法&#xff0c…

151 shell编程

一&#xff0c;我们在终端窗口敲入一个date&#xff0c;系统会返回我们一个结果&#xff0c;这个流程是怎么样子的呢&#xff1f; hunandedehunandede-virtual-machine:~$ date 2024年 03月 24日 星期日 20:32:12 CST当前终端窗口就是 黑屏幕 当我们将date 写入里面并回车的时…

YoloV8改进策略:BackBone改进|PKINet

摘要 PKINet是面向遥感旋转框的主干,网络包含了CAA、PKI等模块,给我们改进卷积结构的模型带来了很多启发。本文,使用PKINet替代YoloV8的主干网络,实现涨点。PKINet是我在作者的模型基础上,重新修改了底层的模块,方便大家轻松移植到YoloV8上。 论文:《Poly Kernel Ince…

C 从函数返回数组

C 语言不允许返回一个完整的数组作为函数的参数。但是&#xff0c;您可以通过指定不带索引的数组名来返回一个指向数组的指针。 如果您想要从函数返回一个一维数组&#xff0c;您必须声明一个返回指针的函数&#xff0c;如下&#xff1a; int * myFunction() { . . . }另外&a…

计算机三级网络技术 选择+大题234笔记

上周停去准备计算机三级的考试啦&#xff0c;在考场上看到题目就知道这次稳了&#xff01;只有一周的时间&#xff0c;背熟笔记&#xff0c;也能稳稳考过计算机三级网络技术&#xff01;