对spring task和线程池的深入研究
目录
- spring task和线程池的研究
- 1、如何实现spring task定时任务的配置
- 2、task里面的一个job方法如何使用多线程,配置线程池
- spring 线程池配置
- 默认线程池ThreadPoolTaskExecutor配置
- 自定义线程池ThreadPoolTaskExecutor配置
spring task和线程池的研究
最近因工作需求,研究了一下spring task定时任务,和线程池,有了一定收获,记录一下
涉及如下内容
1、如何实现spring task定时任务的配置
2、task里面的一个job方法如何使用多线程,配置线程池
如何配置等待子线程结束后,再结束主线程
1、如何实现spring task定时任务的配置
因工作需要,需要定时执行一个方法,通过相关比较后,发现spring自带的task 可以满足,配置简单
步骤
1)增加配置文件 ,在applicationContext-cfg.xml 主配置文件里面添加 相关task标签
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:task="http://www.springframework.org/schema/task" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:jee="http://www.springframework.org/schema/jee" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-3.0.xsd">
2)编写bean类和执行方法
编写jobService类,里面实现testjobThread方法,调用的spring注入过的action、service方法
@Component("jobService") public class jobService { private static Logger logger = Logger.getLogger(jobService.class); @Autowired private ThreadPoolTaskExecutor taskExecutor; final CountDownLatch countDownLatch = new CountDownLatch(3); /** * @Title: DZFP_job * @Description:开票定时任务 */ public void testjobThread() { Date startdate = new Date(); logger.info("DZFP_job_JOB 开始执行任务...,时间 " + startdate); try { DzfpAction.Dzfp_SendAll(); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); logger.error(StringUtil.grabExceptionMessage(e)); } Date enddate = new Date(); logger.info("DZFP_job_JOB 任务完成...时间 " + enddate + " 耗时 " + String.valueOf(enddate.getTime() - startdate.getTime()) + "毫秒"); }
3)配置task相关配置文件,在文件applicationContext-cfg.xml 中增加下列内容
pool-size="5" 该参数主要解决,多个调度并行的问题,如下图5个task任务,建议设置3--5个调度
如果配置参数为 1,下面5个task任务会依次执行,如果一个时间超出,后面的任务一直在等待,影响业务
<!-- 定时任务 --> <task:scheduler id="scheduler" pool-size="5" /> <task:scheduled-tasks scheduler="scheduler"> <!-- 每天7点到7点55, 每隔5分钟执行一次 "0 0/5 7 * * ?"--> <task:scheduled ref="jobService" method="DZFPgetInvoie_job" cron="0 0/30 * * * ?" /> <task:scheduled ref="jobService" method="DZFPgetInvoie_hong_job" cron="0 0/30 * * * ?" /> <task:scheduled ref="jobService" method="testjobThread" cron="0/5 * * * * ?" /> <task:scheduled ref="jobService" method="hzgd_job" cron="0/30 * * * * ?" /> <task:scheduled ref="jobService" method="alipay_pay_job" cron="0/30 * * * * ?" /> </task:scheduled-tasks>
使用以上配置后,启动项目就可以定时执行testjobThread方法里面的业务了。
2、task里面的一个job方法如何使用多线程,配置线程池
经过测试,spring task里面的方法是被串行执行的,比如上面配置的方法 testjobThread方法,5秒执行一次,如果有一个执行过程时间过长,后面的一次调度一直等上次执行结束后,才会启动下一次调用。
也就是说spring task是会监控 执行方法的主线程,如果主线程未结束的话,下一次就不会执行。
根据业务需求,这个testjobThread里面的 业务,需要多线程执行 (批量抽取数据)
spring框架里面,推荐使用线程池
1)配置线程池
在applicationContext-cfg.xml文件中增加配置如下
<!-- spring线程池--> <bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"> <!-- 线程池维护线程的最少数量 --> <property name="corePoolSize" value="5" /> <!-- 线程池维护线程所允许的空闲时间,默认为60s --> <property name="keepAliveSeconds" value="200" /> <!-- 线程池维护线程的最大数量 --> <property name="maxPoolSize" value="20" /> <!-- 缓存队列最大长度 --> <property name="queueCapacity" value="20" /> <!-- 对拒绝task的处理策略 线程池对拒绝任务(无线程可用)的处理策略,目前只支持AbortPolicy、CallerRunsPolicy;默认为后者--> <property name="rejectedExecutionHandler"> <!-- AbortPolicy:直接抛出java.util.concurrent.RejectedExecutionException异常 --> <!-- CallerRunsPolicy:主线程直接执行该任务,执行完之后尝试添加下一个任务到线程池中,可以有效降低向线程池内添加任务的速度 --> <!-- DiscardOldestPolicy:抛弃旧的任务、暂不支持;会导致被丢弃的任务无法再次被执行 --> <!-- DiscardPolicy:抛弃当前任务、暂不支持;会导致被丢弃的任务无法再次被执行 --> <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" /> </property> <property name="waitForTasksToCompleteOnShutdown" value="true" /> </bean>
2)修改业务操作类为thread类,实现run()方法
添加计数器CountDownLatch ,控制子线程结束后,再结束主线程
注意对象实现@Scope("prototype"),用到了成员变量参数
package cn.hao24.action; import java.util.Date; import java.util.concurrent.CountDownLatch; import org.springframework.context.annotation.Scope; import org.springframework.stereotype.Component; import cn.hao24.util.DateUtil; import cn.hao24.util.SpringContextUtils; @Component("testThreadAction") @Scope("prototype") public class testThreadAction extends Thread { /** * spring tash默认是单线程 串行执行,即一个方法执行完成前,后面的job不会执行的 * 但是如果主方法里面产生了thread线程, 主线程如果不等子线程结束后 就结束的话, task任务会产生多次调度 */ private String Treadname; private CountDownLatch latch; public testThreadAction(String Treadname,CountDownLatch latch){ this.Treadname=Treadname; this.latch=latch; } @Override public void run() { try { //主业务方法 for (int i = 0; i < 10; i++) { Thread current = Thread.currentThread(); System.out.println("线程号:"+current.getId() +"--"+current.getName()+" --"+Treadname +":---runing--- "+i+"--"+DateUtil.format(new Date(), "yyyyMMddHHmmss") ); Thread.sleep(20000); } } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); }finally{ //设置实例 执行完毕 latch.countDown(); } } public void setTreadname(String treadname) { Treadname = treadname; } public void setLatch(CountDownLatch latch) { this.latch = latch; } }
2)修改job调度的方法为多线程,配置3个线程
package cn.hao24.job; import java.util.Date; import java.util.concurrent.CountDownLatch; import javax.annotation.Resource; import org.apache.log4j.Logger; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import cn.hao24.action.DzfpAction; import cn.hao24.action.HzgdAction; import cn.hao24.action.KJGOrderjob; import cn.hao24.action.testThreadAction; import cn.hao24.service.ZFBService; import cn.hao24.util.SpringContextUtils; import cn.hao24.util.StringUtil; @Component("jobService") public class jobService { private static Logger logger = Logger.getLogger(jobService.class); @Autowired private ThreadPoolTaskExecutor taskExecutor; final CountDownLatch countDownLatch = new CountDownLatch(3); public void testjobThread() { try { CountDownLatch latch=new CountDownLatch(3); //java工具类,类似与计数器,主要实现子线程未结束钱,主线程一直等待 testThreadAction test1 = (testThreadAction)SpringContextUtils.getBean("testThreadAction","test1",latch); testThreadAction test2 = (testThreadAction)SpringContextUtils.getBean("testThreadAction","test2",latch); testThreadAction test3 = (testThreadAction)SpringContextUtils.getBean("testThreadAction","test3",latch); taskExecutor.execute(test1); taskExecutor.execute(test2); taskExecutor.execute(test3); latch.await(); //子线程未结束前,一直等待 //test1.run(); } catch (Exception e) { e.printStackTrace(); logger.error(StringUtil.grabExceptionMessage(e)); } } }
执行效果如下:
虽然 testjobThread 5秒执行一次,但是因为使用到了 latch.await() latch.countDown();需要等子线程执行完毕,才会进行下一次job
子线程每次循环,会sleep 20秒,从下面结果看,3个线程 每隔20秒才打印一次。符合最终要求
线程号:29--taskExecutor-3 --test3:---runing--- 0--20170622145500
线程号:28--taskExecutor-2 --test2:---runing--- 0--20170622145500
线程号:27--taskExecutor-1 --test1:---runing--- 0--20170622145500
线程号:28--taskExecutor-2 --test2:---runing--- 1--20170622145520
线程号:27--taskExecutor-1 --test1:---runing--- 1--20170622145520
线程号:29--taskExecutor-3 --test3:---runing--- 1--20170622145520
线程号:29--taskExecutor-3 --test3:---runing--- 2--20170622145540
线程号:28--taskExecutor-2 --test2:---runing--- 2--20170622145540
线程号:27--taskExecutor-1 --test1:---runing--- 2--20170622145540
spring 线程池配置
默认线程池ThreadPoolTaskExecutor配置
配置核心参数
直接在application.properties中配置核心参数
spring.task.execution.pool.core-size=8 spring.task.execution.pool.max-size=12 spring.task.execution.pool.keep-alive=60s spring.task.execution.pool.queue-capacity=100000 spring.task.execution.pool.allow-core-thread-timeout=true spring.task.execution.thread-name-prefix=swy-task-
创建JavaBean注入
@Configuration public class ExecutorConfig { private static final Logger logger = LoggerFactory.getLogger(ExecutorConfig.class); @Bean public Executor asyncServiceExecutor() { logger.info("start asyncServiceExecutor"); ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); //配置核心线程数 executor.setCorePoolSize(5); //配置最大线程数 executor.setMaxPoolSize(6); //配置队列大小 executor.setQueueCapacity(99999); //配置线程池中的线程的名称前缀 executor.setThreadNamePrefix("swy-task-"); // rejection-policy:当pool已经达到max size的时候,如何处理新任务 // CALLER_RUNS:不在新线程中执行任务,而是有调用者所在的线程来执行 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); //执行初始化 executor.initialize(); return executor; } }
在配置类,或入口类开启@EnableAsync注解
@SpringBootApplication @EnableAsync public class MultiThreadApplication { public static void main(String[] args) { SpringApplication.run(MultiThreadApplication.class, args); } }
在Service层或Controller层的类或方法上添加@Async注解
@Async public void doSomethingAsync(){ logger.info("start executeAsync"); try{ Thread.sleep(5000); }catch(Exception e){ e.printStackTrace(); } logger.info("end executeAsync"); }
自定义线程池ThreadPoolTaskExecutor配置
继承ThreadPoolTaskExecutor创建新线程池类
public class CustomThreadPoolTaskExecutor extends ThreadPoolTaskExecutor { private static final Logger logger = LoggerFactory.getLogger(CustomThreadPoolTaskExecutor.class); private void showThreadPoolInfo(String prefix){ ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor(); if(null==threadPoolExecutor){ return; } logger.info("{}, {},taskCount [{}], completedTaskCount [{}], activeCount [{}], queueSize [{}]", this.getThreadNamePrefix(), prefix, threadPoolExecutor.getTaskCount(), threadPoolExecutor.getCompletedTaskCount(), threadPoolExecutor.getActiveCount(), threadPoolExecutor.getQueue().size()); } @Override public void execute(Runnable task) { showThreadPoolInfo("1. do execute"); super.execute(task); } @Override public void execute(Runnable task, long startTimeout) { showThreadPoolInfo("2. do execute"); super.execute(task, startTimeout); } @Override public Future<?> submit(Runnable task) { showThreadPoolInfo("1. do submit"); return super.submit(task); } @Override public <T> Future<T> submit(Callable<T> task) { showThreadPoolInfo("2. do submit"); return super.submit(task); } @Override public ListenableFuture<?> submitListenable(Runnable task) { showThreadPoolInfo("1. do submitListenable"); return super.submitListenable(task); } @Override public <T> ListenableFuture<T> submitListenable(Callable<T> task) { showThreadPoolInfo("2. do submitListenable"); return super.submitListenable(task); } }
配置新建线程池类的核心参数
@Configuration public class ExecutorConfig { private static final Logger logger = LoggerFactory.getLogger(ExecutorConfig.class); @Bean public Executor asyncServiceExecutor() { logger.info("start asyncServiceExecutor"); ThreadPoolTaskExecutor executor = new CustomThreadPoolTaskExecutor(); //配置核心线程数 executor.setCorePoolSize(5); //配置最大线程数 executor.setMaxPoolSize(8); //配置队列大小 executor.setQueueCapacity(99999); //配置线程池中的线程的名称前缀 executor.setThreadNamePrefix("async-service-"); // rejection-policy:当pool已经达到max size的时候,如何处理新任务 // CALLER_RUNS:不在新线程中执行任务,而是有调用者所在的线程来执行 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); //执行初始化 executor.initialize(); return executor; } }
在配置类,或入口类开启@EnableAsync注解
@SpringBootApplication @EnableAsync public class MultiThreadApplication { public static void main(String[] args) { SpringApplication.run(MultiThreadApplication.class, args); } }
在Service层或Controller层的类或方法上添加@Async注解,此时需需注意一定要注明Bean方法名称。
@Async("asyncServiceExecutor") public void doSomethingAsync(){ logger.info("start executeAsync"); try{ Thread.sleep(5000); }catch(Exception e){ e.printStackTrace(); } logger.info("end executeAsync"); }
以上为个人经验,希望能给大家一个参考,也希望大家多多支持我们。