JUC回顾之-ThreadPoolExecutor的原理和使用
Spring中的ThreadPoolTaskExecutor是借助于JDK并发包中的java.util.concurrent.ThreadPoolExecutor来实现的。基于ThreadPoolExecutor可以很容易将一个Runnable接口的任务放入线程池中。
ThreadPoolExecutor的构建参数:
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);}
1. 参数解释
corePoolSize: 核心线程数,会一直存活,即使没有任务,线程池也会维护线程的最少数量
maximumPoolSize: 线程池维护线程的最大数量
keepAliveTime: 线程池维护线程所允许的空闲时间,当线程空闲时间达到keepAliveTime,该线程会退出,直到线程数量等于corePoolSize。如果allowCoreThreadTimeout设置为true,则所有线程均会退出直到线程数量为0。
unit: 线程池维护线程所允许的空闲时间的单位、可选参数值为:TimeUnit中的几个静态属性:NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDS。
workQueue: 线程池所使用的缓冲队列,常用的是:java.util.concurrent.ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue
解释:
用来保存等待被执行的任务的阻塞队列,且任务必须实现Runable接口,在JDK中提供了如下阻塞队列:
- ArrayBlockingQueue:基于数组结构的有界阻塞队列,按FIFO排序任务;
- LinkedBlockingQuene:基于链表结构的阻塞队列,按FIFO排序任务,吞吐量通常要高于ArrayBlockingQuene;
- SynchronousQuene:一个不存储元素的阻塞队列,每个插入操作必须等到另一个线程调用移除操作,否则插入操作一直处于阻塞状态,吞吐量通常要高于LinkedBlockingQuene;
- priorityBlockingQuene:具有优先级的无界阻塞队列;
handler: 线程池中的数量大于maximumPoolSize,对拒绝任务的处理策略,默认值ThreadPoolExecutor.AbortPolicy()。
解释:
线程池的饱和策略,当阻塞队列满了,且没有空闲的工作线程,如果继续提交任务,必须采取一种策略处理该任务,线程池提供了4种策略:
- AbortPolicy:直接抛出异常,默认策略;
- CallerRunsPolicy:用调用者所在的线程来执行任务;
- DiscardOldestPolicy:丢弃阻塞队列中靠最前的任务,并执行当前任务;
- DiscardPolicy:直接丢弃任务;
当然也可以根据应用场景实现RejectedExecutionHandler接口,自定义饱和策略,如记录日志或持久化存储不能处理的任务。
2.ThreadPoolExecutor内运转机制:
具体流程如下:当一个任务通过execute(Runnable)方法欲添加到线程池时:
- 当池子大小小于corePoolSize就新建线程,并且处理请求。
- 当池子大小等于corePoolSize,把请求放入workQueue中,池子里的空闲线程就去从workQueue中取任务并处理。
- 当workQueue放不下新任务时,新建线程入池,并处理请求,如果池子的大小撑到maximumPoolSize就用RejectedExecutionHandler拒绝处理。
- 另外,当池子的线程数大于corePoolSize的时候,多余的线程会等待keepAliveTime长的时间,如果无请求可处理就自行销毁。
3.Exectors
Exectors工厂类提供了线程池的初始化接口,主要有如下几种:
newFixedThreadPool
初始化一个指定线程数的线程池,其中corePoolSize == maximumPoolSize,使用LinkedBlockingQuene作为阻塞队列,不过当线程池没有可执行任务时,也不会释放线程。
注意:由于LinkedBlockingQuene队列的大小是整型的最大值,队列是无界队列,所以不建议直接使用,newFixedThreadPool可能导致内存爆满的问题;
newCachedThreadPool
1、初始化一个可以缓存线程的线程池,默认缓存60s,线程池的线程数可达到Integer.MAX_VALUE,即2147483647,内部使用SynchronousQueue作为阻塞队列;
2、和newFixedThreadPool创建的线程池不同,newCachedThreadPool在没有任务执行时,当线程的空闲时间超过keepAliveTime,会自动释放线程资源,当提交新任务时,如果没有空闲线程,则创建新线程执行任务,会导致一定的系统开销;
注意:使用该线程池时,一定要注意控制并发的任务数,因为最大的线程数可以达到整型的最大值,线程数是无限的,当并发的任务很多时候,会同时创建大量的线程,会导致严重的性能问题。
newSingleThreadExecutor
初始化的线程池中只有一个线程,如果该线程异常结束,会重新创建一个新的线程继续执行任务,唯一的线程可以保证所提交任务的顺序执行,内部使用LinkedBlockingQueue作为阻塞队列。
注意:由于LinkedBlockingQuene队列的大小是整型的最大值,队列是无界队列,所以不建议直接使用,newSingleThreadExecutor可能导致内存爆满的问题;
newScheduledThreadPool
初始化的线程池可以在指定的时间内周期性的执行所提交的任务,在实际的业务场景中可以使用该线程池定期的同步数据。
3.spingMVC中使用线程池,用做批处理
<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"xmlns:mvc="http://www.springframework.org/schema/mvc" xmlns:aop="http://www.springframework.org/schema/aop"xmlns:tx="http://www.springframework.org/schema/tx" xmlns:oscache="http://www.springmodules.org/schema/oscache"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsdhttp://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-3.0.xsdhttp://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsdhttp://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsdhttp://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsdhttp://www.springmodules.org/schema/oscache http://www.springmodules.org/schema/cache/springmodules-oscache.xsd"><!-- spring 加载资源文件的配置 --><context:property-placeholderlocation="classpath:conf/custom/env/config.properties"ignore-unresolvable="true" /><!-- spring 线程池配置 start --><!-- insurance线程池 --><bean id="insuranceTaskExecutor"class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"><property name="corePoolSize" value="${insurance.taskexecutor.corePoolSize}" /><property name="maxPoolSize" value="${insurance.taskexecutor.maxPoolSize}" /><property name="keepAliveSeconds" value="${insurance.taskexecutor.keepAliveSeconds}" /><property name="queueCapacity" value="${insurance.taskexecutor.queueCapacity}" /></bean></beans>
config里面对于线程池的配置:
#insurance task setting insurance.taskexecutor.corePoolSize=2 insurance.taskexecutor.maxPoolSize=2 insurance.taskexecutor.keepAliveSeconds=30 insurance.taskexecutor.queueCapacity=1000
Controller 层部分代码:
/*** * sendInsuranceBatch* * @Title: sendInsuranceBatch* @Description: TODO 投保批处理* @param pojo* @param errors* @param request* @return*/@RequestMapping("/ihotel_insurance_deliver")@ResponseBodypublic String sendInsuranceBatch(@ModelAttribute("pojo") IhotelBatchPojo pojo, Errors errors,HttpServletRequest request) {String remoteIp = WebUtils.getRemoteIpAddress(request);String serverIP = IPUtil.getServerIp();BaseResultInfo baseResultInfo = null;// 验证入参是否正确try {// 1,校验参数 pojo.validate(pojo, errors);if (errors.hasErrors()) {baseResultInfo = new BaseResultInfo(ResultType.PARAMETER_VERIFY_FAILUE);baseResultInfo.appendRetdesc("#" + getErrMsg(errors));return JsonUtil.BeanToJson(baseResultInfo);}ihotelInsuranceBatchService.processIhotelInsuredItem(pojo, remoteIp);baseResultInfo = new BaseResultInfo();baseResultInfo.setRetcode(0);baseResultInfo.setRetdesc("success");baseResultInfo.setServerIP(serverIP);return JSON.toJSONString(baseResultInfo);} catch (Exception e) {IHotelLoggerUtil.error("[国际酒店保险投保批处理异常]", e);throw e;}}
Service层部分代码:
@Service("ihotelInsuranceBatchService") public class IhotelInsuranceBatchService {// 保险线程池 @Resourceprivate ThreadPoolTaskExecutor insuranceTaskExecutor;public void processIhotelInsuredItem(IhotelBatchPojo pojo, String remoteIp) {// 根据入参条件取出要投保批处理的数据List<InsuredHotelItem> isuredHotelList = this.findListByCond(pojo);for (InsuredHotelItem ihotelItem : isuredHotelList) {//将任务放入线程池insuranceTaskExecutor.execute(new IhotelInsureDeliverTask(ihotelItem, this, remoteIp));}}}
线程类
/*** @Title: IhotelInsureDeliverTask.java* @Package com.elong.ihotel.service.insurance* @Description: TODO* Copyright: Copyright (c) 2014 * Email: songbin0819@163.com* * @author user* @date 2014年11月7日 下午4:04:02* @version V1.0*/package com.elong.ihotel.service.insurance;import com.elong.ihotel.model.insurance.InsuredHotelItem; import com.elong.ihotel.util.IHotelLoggerUtil;/*** IhotelInsureDeliverTask* * @Title: IhotelInsureDeliverTask* @Description: TODO 保险投保批处理线程类* @author Peng.Li* @date 2014年11月7日 下午4:04:02**/public class IhotelInsureDeliverTask implements Runnable {/*** 需要批处理的保险中间表实体*/private InsuredHotelItem ihotelItem;private IhotelInsuranceBatchService ihotelInsuranceBatchService;/*** ip*/private String remoteIp;/*** Constructor for IhotelInsureDeliverTask.* <p>* Title:* </p>* <p>* Description:* </p>*/public IhotelInsureDeliverTask() {}/*** Constructor for IhotelInsureDeliverTask.* <p>* Title:* </p>* <p>* Description:* </p>* * @param ihotelItem* @param ihotelInsuranceBatchService* @param remoteIp*/public IhotelInsureDeliverTask(InsuredHotelItem ihotelItem,IhotelInsuranceBatchService ihotelInsuranceBatchService, String remoteIp) {this.ihotelItem = ihotelItem;this.ihotelInsuranceBatchService = ihotelInsuranceBatchService;this.remoteIp = remoteIp;}/*** * override run* <p>* Title: run* </p>* <p>* Description:* </p>*/@Overridepublic void run() {try {ihotelInsuranceBatchService.sendInsuranceBatch(ihotelItem, remoteIp);} catch (Exception e) {IHotelLoggerUtil.error("获取保险确认号失败", e);}}}
转载于:https://www.cnblogs.com/200911/p/4309512.html
JUC回顾之-ThreadPoolExecutor的原理和使用相关推荐
- JDK ThreadPoolExecutor核心原理与实践
作者:vivo互联网服务器团队-Xu Weiteng 一.内容概括 本文内容主要围绕JDK中的ThreadPoolExecutor展开,首先描述了ThreadPoolExecutor的构造流程以及内部 ...
- JAVA8线程池THREADPOOLEXECUTOR底层原理及其源码解析
小侃一下 1. 使用线程池的好处. 为什么要使用线程池? 2. 线程池核心参数介绍 3. 提交任务到线程池中的流程 3.1 ThreadPoolExecutor#execute方法整体流程 3.2 排 ...
- C++线程池ThreadPoolExecutor实现原理
1. 为什么要使用线程池 在实际使用中,线程是很占用系统资源的,如果对线程管理不善很容易导致系统问题.因此,在大多数并发框架中都会使用线程池来管理线程,使用线程池管理线程主要有如下好处: 降低资源消耗 ...
- Java并发系列(11)——ThreadPoolExecutor实现原理与手写
接上一篇<Java并发系列(10)--FutureTask 和 CompletionService> 文章目录 9 线程池 9.1 JDK 线程池 9.2 ThreadPoolExecut ...
- Java线程池:ThreadPoolExecutor运行原理
ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 2, 3, TimeUnit.SECONDS, new LinkedBlockingQueue& ...
- Java多线程系列--“JUC锁”10之 CyclicBarrier原理和示例
CyclicBarrier简介 CyclicBarrier是一个同步辅助类,允许一组线程互相等待,直到到达某个公共屏障点 (common barrier point).因为该 barrier 在释放等 ...
- Java 并发编程-不懂原理多吃亏(送书福利)
作者 | 加多 关注阿里巴巴云原生公众号,后台回复关键字"并发",即可参与送书抽奖! ** 导读:并发编程与 Java 中其他知识点相比较而言学习门槛较高,从而导致很多人望而却步. ...
- java线程池使用详解ThreadPoolExecutor使用示例
一 使用线程池的好处 二 Executor 框架 2.1 简介 2.2 Executor 框架结构(主要由三大部分组成) 1) 任务(Runnable /Callable) 2) 任务的执行(Exec ...
- 13.ThreadPoolExecutor线程池之submit方法
jdk1.7.0_79 在上一篇<ThreadPoolExecutor线程池原理及其execute方法>中提到了线程池ThreadPoolExecutor的原理以及它的execute方法 ...
最新文章
- arm服务器获取文件路径中文,ssh 访问远程服务器文件路径
- TensorFlow数据读取方式:Dataset API,以及如何查看dataset:DatasetV1Adapter的方法
- Zabbix 监控Mysql 主从状态(Python脚本)
- hybriddb mysql移植_HybridDB for MySQL 实现在线与离线数据分离的实践
- 【翻译】【linux设备驱动】linux地址类型
- N76E003介绍与开发资料获取
- 如何使用Java与Mysql进行数据交互
- 04-程序计数器(PC计数器)
- SAP Spartacus里的converter实例化逻辑
- 唤醒计算机运行此任务_如何停止Windows 8唤醒计算机以运行维护
- elaselasticsearch节点类型
- 新iPhone XR全配色曝光:苹果你清醒一点 还在搞配色?
- 笑话(12) 内裤都磨破了
- 【Python实例第28讲】核主成分
- 【日常折腾】Y7000P拆机-更换触控板
- 用AVX2指令集优化整形数组求和
- python实现阿里云盘同步功能(sync_folder)
- Flink 极简教程: 架构及原理 Apache Flink® — Stateful Computations over Data Streams
- 记一次Windows 无法加载这个硬件的设备驱动程序。驱动程序可能已损坏或不见了。 (代码 39)
- HTML 5入门基础
热门文章
- Debian的版本一些记录-sid, testing, buster, bullseye, stretch,jessie
- MYSQL 单表删除1000万数据
- Echarts中series、option、component究竟是啥?
- Android代码修改SIM的PIN码,中兴U970如何修改SIM卡或UIM卡的PIN码
- 嵌入式linux设备驱动开发,嵌入式Linux设备驱动开发简介.pdf
- Qt5应用内嵌输入法(支持中文拼音输入、英文输入、数字输入、常用符号输入、手写输入、实体键盘同步)
- chrome真机调试Android
- DELMIA软件 机器人IK的设定
- hp 服务器可以安装win7系统安装失败,HP笔记本UEFI导致无法安装win7系统的解决方法...
- # Java简易计算器编程思想