浅谈线程池的拓容、判断、限速技巧

news/2024/7/7 22:35:45

本文主要分享线程池的动态拓容、判断执行任务执行完毕、线程任务提交速度限制三个线程池使用技巧。

线程池动态拓容

创建线程池

ThreadPoolExecutor taskExecutor = new ThreadPoolExecutor(8, 64, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<>(256), new ThreadFactoryBuilder().setNameFormat("customer-worker-%d").build(), (task, executor) -> { try { executor.getQueue().put(task); } catch (InterruptedException e) { logger.error("customer-worker 线程阻断异常:" + e.getMessage(), e); } }); 复制代码

动态拓展容量

触发方式，我们可以通过 apollo 配置或者通过延迟任务定期扫描自动拓展

// 配置信息 @Value("${corePoolSize:2}") private Integer corePoolSize; @Value("${corePoolSize:8}") private Integer maximumPoolSize; // 修改线程池核心参数，提高并发度 taskExecutor.setCorePoolSize(corePoolSize); taskExecutor.setMaximumPoolSize(maximumPoolSize); 复制代码

可以通过这种方式对生产环境线程池参数进行动态优化。

判断线程池执行结束

线程池的使用可以提高我们并发程序线程复用，以及提供对线程的管理能力。提高线程的利用率，提升性能。但是对于一些场景我们需要知道当前线程池中提交的任务是否执行完毕，我们可以通过以下4种方式判断。

使用 isTerminated 方法

threadPool.isTerminated() 常用来判断线程池是否结束，结束了为TRUE.

使用 threadPool.isTerminated() 方法，必须在shutdown()方法关闭线程池之后才能使用，否则isTerminated()永不为TRUE，线程将一直阻塞在该判断的地方，导致程序最终崩溃。

/** * 线程池任务执行完成判断 */ public class ThreadPoolCompleted { public static void main(String[] args) { // 1.创建线程池 ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20, 0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1024)); // 2.添加任务 addTask(threadPool); // 3.判断线程池是否执行完 isCompleted(threadPool); // 【核心调用方法】 // 4.线程池执行完 System.out.println(); System.out.println("线程池任务执行完成！"); } /** * 方法1：isTerminated 实现方式 * 判断线程池的所有任务是否执行完 */ private static void isCompleted(ThreadPoolExecutor threadPool) { threadPool.shutdown(); while (!threadPool.isTerminated()) { // 如果没有执行完就一直循环 } } /** * 给线程池添加任务 */ private static void addTask(ThreadPoolExecutor threadPool) { // 任务总数 final int taskCount = 5; // 添加任务 for (int i = 0; i < taskCount; i++) { final int finalI = i; threadPool.submit(new Runnable() { @Override public void run() { try { // 随机休眠 0-4s int sleepTime = new Random().nextInt(5); TimeUnit.SECONDS.sleep(sleepTime); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(String.format("任务%d执行完成", finalI)); } }); } } } 复制代码

使用 getCompletedTaskCount 方法

我们可以通过判断线程池中的计划执行任务数和已完成任务数，来判断线程池是否已经全部执行完，如果计划执行任务数=已完成任务数，那么线程池的任务就全部执行完了，否则就未执行完。

/** * 方法2：getCompletedTaskCount 实现方式 * 判断线程池的所有任务是否执行完 */ private static void isCompletedByTaskCount(ThreadPoolExecutor threadPool) { while (threadPool.getTaskCount() != threadPool.getCompletedTaskCount()) { } } 复制代码

方法解析：

  • getTaskCount()：返回计划执行的任务总数。由于任务和线程的状态可能在计算过程中动态变化，因此返回的值只是一个近似值。

  • getCompletedTaskCount()：返回完成执行任务的总数。因为任务和线程的状态可能在计算过程中动态地改变，所以返回的值只是一个近似值，但是在连续的调用中并不会减少。

PS: 此实现方法的优点是无需关闭线程池。 它的缺点是getTaskCount() 和getCompletedTaskCount(),返回的是一个近似值，因为线程池中的任务和线程的状态可能在计算过程中动态变化，所以它们两个返回的都是一个近似值。

使用 CountDownLatch

CountDownLatch 是一个栅栏类似计数器，我们创建了一个包含 N 个任务的计数器，每个任务执行完计数器 -1，直到计数器减为 0 时，说明所有的任务都执行完了

//创建线程池 ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20, 0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1024)); CountDownLatch countDownLatch = new CountDownLatch(2); for (int i = 0; i < 10; i++) { int finalI = i; threadPool.submit(() -> { try { // 随机休眠 0-4s int sleepTime = new Random().nextInt(5); TimeUnit.SECONDS.sleep(Math.max(sleepTime, 2)); } catch (InterruptedException e) { System.err.println("InterruptedException err : " + e.getMessage()); } System.out.println(gettime() + " task running : " + finalI); countDownLatch.countDown(); // 每次计数器-1 }); } try { countDownLatch.await(); // 等待所有的任务执行完毕 } catch (InterruptedException t) { System.err.println("InterruptedException err : " + t.getMessage()); } 复制代码

CountDownLatch 写法很优雅，且无需关闭线程池，但它的缺点是只能使用一次，CountDownLatch创建之后不能被重复使用，也就是说 CountDownLatch 可以理解为只能使用一次的计数器。

使用 CyclicBarrier

CyclicBarrier 和 CountDownLatch 类似，它可以理解为一个可以重复使用的循环计数器，CyclicBarrier 可以调用 reset 方法将自己重置到初始状态，CyclicBarrier 具体实现代码如下：

// 创建线程池 ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20, 0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1024)); final int taskCount = 5; // 任务总数 // 循环计数器 CyclicBarrier cyclicBarrier = new CyclicBarrier(taskCount, new Runnable() { @Override public void run() { // 线程池执行完 System.out.println(); System.out.println("线程池所有任务已执行完！"); } }); // 添加任务 for (int i = 0; i < taskCount; i++) { final int finalI = i; threadPool.submit(new Runnable() { @Override public void run() { try { // 随机休眠 0-4s int sleepTime = new Random().nextInt(5); TimeUnit.SECONDS.sleep(sleepTime); System.out.println(String.format("任务%d执行完成", finalI)); // 线程执行完 cyclicBarrier.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } } }); } 复制代码

执行结果：

​CyclicBarrier 构造方法核心参数:

  • parties 计数器的数量 ，

  • barrierAction 计数器为 0 时，也就是任务都执行完之后可以执行的事件（方法）。

await 方法：

在 CyclicBarrier 上进行阻塞等待，当调用此方法时 CyclicBarrier 的内部计数器会 -1，直到发生以下情形之一：

在 CyclicBarrier 上等待的线程数量达到 parties，也就是计数器的声明数量时，则所有线程被释放，继续执行。

当前线程被中断，则抛出 InterruptedException 异常，并停止等待，继续执行。

其他等待的线程被中断，则当前线程抛出 BrokenBarrierException 异常，并停止等待，继续执行。

其他等待的线程超时，则当前线程抛出 BrokenBarrierException 异常，并停止等待，继续执行。

其他线程调用 CyclicBarrier.reset() 方法，则当前线程抛出 BrokenBarrierException 异常，并停止等待，继续执行。

reset 方法：

使得CyclicBarrier回归初始状态，直观来看它做了两件事：

如果有正在等待的线程，则会抛出 BrokenBarrierException 异常，且这些线程停止等待，继续执行。

将是否破损标志位 broken 置为 false。

优缺点分析

CyclicBarrier 从设计的复杂度到使用的复杂度都高于 CountDownLatch，相比于 CountDownLatch 来说它的优点是可以重复使用（只需调用 reset 就能恢复到初始状态），缺点是使用难度较高。

4 种方法总结:

  1. 使用 isTerminated 方法判断：通过判断线程池的完成状态来实现，需要关闭线程池，一般情况下不建议使用。

  2. 使用 getCompletedTaskCount 方法判断：通过计划执行总任务量和已经完成总任务量，来判断线程池的任务是否已经全部执行，如果相等则判定为全部执行完成。但因为线程个体和状态都会发生改变，所以得到的是一个大致的值，可能不准确。

  3. 使用 CountDownLatch 判断：相当于一个线程安全的单次计数器，使用比较简单，且不需要关闭线程池，是比较常用的判断方法。

  4. 使用 CyclicBarrier 判断：相当于一个线程安全的重复计数器，但使用较为复杂，所以日常项目中使用的较少。

线程池提交速度限制

对于线程池提交速率限制，我们可以通过信号量进行限制，这样可以避免触发线程池的拒绝策略，将逻辑处理能力限制功能前置，可以做到线程池的职责单一，负责提交而去做额外的事情。

下面是一个简单的例子：

// 创建线程池 ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20, 0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1024)); final int taskCount = 5; // 任务总数 // 循环计数器 Semaphore semaphore = new Semaphore(2); // 添加任务 for (int i = 0; i < taskCount; i++) { final int finalI = i; threadPool.submit(new Runnable() { @Override public void run() { try { // 获取信号量 semaphore.acquire(); // 随机休眠 4s TimeUnit.SECONDS.sleep(4); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); String format1 = sdf.format(new Date()); System.out.println(String.format("%s 任务%d执行完成", format1, finalI)); } catch (InterruptedException e) { e.printStackTrace(); } finally { semaphore.release(); } } }); } 复制代码

执行结果：

​我们可以看到，我们同一个时间只能获取两个信号量，也就是两个线程能进行执行，如果新进入线程调用 semaphore.acquire();方法进入线程阻塞，需要等待有信号量释放过后才能进入新的调用。释放方法 semaphore.release();.


http://lihuaxi.xjx100.cn/news/141996.html

相关文章

Java Spring Cloud XXIII 之 配置中心

Java Spring Cloud XXIII 之 配置中心

Java Spring Cloud XXIII 之 配置中心 配置中心 1.配置中心简介 所谓配置中心:将项目需要的配置信息保存在配置中心,需要读取时直接从配置中心读取,方便配置管理的微服务工具 我们可以将部分yml文件的内容保存在配置中心 一个微服务项目有很多子模块,这些子模块可能在不同…
阅读更多...
中国及全球LNG液化天然气行业发展趋势分析告

中国及全球LNG液化天然气行业发展趋势分析告

智研瞻产业研究院专注于中国产业经济情报及研究&#xff0c;目前主要提供的产品和服务包括传统及新兴行业研究、商业计划书、可行性研究、市场调研、专题报告、定制报告等。涵盖文化体育、物流旅游、健康养老、生物医药、能源化工、装备制造、汽车电子、农林牧渔等领域&#xf…
阅读更多...
hyperf 模型批量更新数据

hyperf 模型批量更新数据

使用hyperf框架没有批量更新方法&#xff0c;只能自己拼接SQL语句进行批量跟你更新 封装一个SQL句更新方法 //批量更新public function updateBatch($multipleData []){try {$tableName Db::getTablePrefix() . table;// 获取到数据里面的所有字段名$firstRow current($m…
阅读更多...
Linux下NANDFLASH probe函数分析

Linux下NANDFLASH probe函数分析

本文记录一下自己平台上NANDFLASH驱动的执行流程。 驱动入口&#xff1a; module_platform_driver(ali_nand_driver); module_platform_driver是一个宏&#xff0c;位于kernel根目录下include/linux/platform_device.h,其展开如下: #define module_platform_driver(__platfo…
阅读更多...
Java中的容器(二) 双例集合

Java中的容器(二) 双例集合

容器(二) 双例集合 1、Map接口介绍 Map中不能包含重复的Key&#xff0c;但可以包含重复的Value。 Map中的常用方法 put方法&#xff1a;key不存在时返回空&#xff0c;key存在时更新value&#xff0c;并返回旧的value。 2、HashMap容器类 通过KeySet获取元素 Set<String…
阅读更多...
VectorDraw开发者框架(VDF)

VectorDraw开发者框架(VDF)

VectorDraw开发者框架(VDF) VectorDraw Developer Framework(VDF)是一个易于创建、管理和打印2D和3D图纸的组件。VectorDraw对象公开了与最常见的矢量格式和其他CAD对象兼容的方法和属性。它支持10多种矢量格式和多种光栅格式。VectorDrawDeveloperFramework(VDF)是完全面向对象…
阅读更多...
arraybuffer 转json

arraybuffer 转json

场景后端返回的数据是arraybuffer 类型&#xff0c;请求成功下载excel&#xff0c;失败弹出错误原因 <script setup lang"ts"> import { ElMessage} from element-plus; // 下载; const exportData async (fileName: string) > {const res await downloa…
阅读更多...
详解Git合并(Merge)错误如何回退。（包括Reset, Revert和页面回滚三种，并说明其优缺点）

详解Git合并(Merge)错误如何回退。（包括Reset, Revert和页面回滚三种，并说明其优缺点）

文章目录1. 问题场景描述1.1 场景模拟2. 解决方案2.1 利用git reset --hard命令2.2 利用git revert 命令2.3 使用页面进行回滚&#xff08;效果与Revert一致&#xff09;1. 问题场景描述 我的项目包含两个重要git分支&#xff1a;master&#xff08;生产环境&#xff09;&…
阅读更多...
最新文章

Copyright @ 2022~2023