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.hqwc.cn/news/564230.html

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

相关文章

mysql80-DBA数据库学习1

掌握能力 核心技能 核心技能 mysql部署 官网地址www.mysql.com 或者www.oracle.com https://dev.mysql.com/downloads/repo/yum/ Install the RPM you downloaded for your system, for example: yum install mysql80-community-release-{platform}-{version-number}.noarch…

嵌入式数据库--SQLite

目录 1. SQLite数据库简介 2. SQLite数据库的安装 方式一&#xff1a; 方式二&#xff1a; 3. SQLite的命令用法 1.创建一个数据库 2.创建一张表 3.删除表 4.插入数据 5. 查询数据 6.删除表内一条数据 7.修改表中的数据 8.增加一列也就是增加一个字段 1. SQLite数据库…

配置OpenFrp安装使用,配置后台启动和开机自启动

首先说明&#xff0c;我的系统是&#xff1a;Ubuntu 安装 创建一个目录用于放置相关文件 mkdir frp进入目录 cd frp下载openfrp客户端 # 如果您使用 wget&#xff0c;用以下命令: wget https://o.of.gs/client/OpenFRP_0.56.0_ce243e7e_20240320/frpc_linux_amd64.tar.gz …

手撕算法-无重复字符的最长子串

描述 分析 滑动窗口&#xff0c;记录窗口中的所有出现的字符&#xff0c;然后窗口左边界固定&#xff0c;右边界右滑&#xff0c;如果&#xff0c;窗口中不存在新的字符&#xff0c;则右滑成功&#xff0c;否则左边界右滑&#xff0c;直到窗口中不存在右边界的值。 描述感觉不…

03-SparkSQL入门

0 Shark Spark 的一个组件&#xff0c;用于大规模数据分析的 SQL 查询引擎。Shark 提供了一种基于 SQL 的交互式查询方式&#xff0c;可以让用户轻松地对大规模数据集进行查询和分析。Shark 基于 Hive 项目&#xff0c;使用 Hive 的元数据存储和查询语法&#xff0c;并基于Hiv…

java每日一题——买啤酒(递归经典问题)

前言&#xff1a; 非常喜欢的一道题&#xff0c;经典中的经典。打好基础&#xff0c;daydayup!!!啤酒问题&#xff1a;一瓶啤酒2元&#xff0c;4个盖子可以换一瓶&#xff0c;2个空瓶可以换一瓶&#xff0c;请问10元可以喝几瓶 题目如下&#xff1a; 啤酒问题&#xff1a;一瓶…

【Linux 驱动基础】IMX6ULL LED基础驱动

本机使用的是正点原子的IMX6ULL开发板 # 前置知识 IMX6ULL GPIO控制框图&#xff1a; GPIO控制代码大概分为几个流程&#xff1a;开启时钟、设置IO复用、设置IO属性、配置IO方向、设置IO输出电平&#xff0c;下面以IMX6ULL为例&#xff0c; 1. 开启时钟 参考资料&#xff1a…

qt Qt Remote Object(QtRO)实现进程间通信

简介 Qt Remote Object简称QtRO&#xff0c;这是Qt5.9以后官方推出来的新模块&#xff0c;专门用于进程间通信&#xff08;IPC&#xff09;。是基于Socket来封装的&#xff0c;兼容LPC和RPC。LPC即Local Process Communication&#xff0c;而RPC是指Remote Process Communicat…

计算机软件安全

一、软件安全涉及的范围 1.1软件本身的安全保密 软件的本质与特征&#xff1a; 可移植性 寄生性 再生性 可激发性 攻击性 破坏性 …… 知识产权与软件盗版 软件商品交易形式不透明&#xff0c;方式多样&#xff0c;传统商标标识方法不适用&#xff1b; 盗版方法简捷…

基于java+springboot+vue实现的研究生志愿填报辅助系统(文末源码+Lw+ppt)23-600

摘 要 二十一世纪我们的社会进入了信息时代&#xff0c;信息管理系统的建立&#xff0c;大大提高了人们信息化水平。传统的管理方式对时间、地点的限制太多&#xff0c;而在线管理系统刚好能满足这些需求&#xff0c;在线管理系统突破了传统管理方式的局限性。于是本文针对这…

Verilog刷题笔记45

题目&#xff1a;Given the finite state machine circuit as shown, assume that the D flip-flops are initially reset to zero before the machine begins. Build this circuit. 解题&#xff1a; module top_module (input clk,input x,output z ); wire [2:0]size;dtou…

CTF-辨别细菌

题目描述&#xff1a;try your best to find the flag. 进入靶场后发现是一个游戏&#xff0c;需要全部答对才可以得到最后的flag 查看了一下源码&#xff0c;发现有一个答案模板的模块 尝试解释一下代码 <!-- 答案模版 --> <script id"template_game_pi…