對spring task和線程池的深入研究
spring task和線程池的研究
最近因工作需求,研究了一下spring task定時任務(wù),和線程池,有了一定收獲,記錄一下
涉及如下內(nèi)容
1、如何實(shí)現(xiàn)spring task定時任務(wù)的配置
2、task里面的一個job方法如何使用多線程,配置線程池
如何配置等待子線程結(jié)束后,再結(jié)束主線程
1、如何實(shí)現(xiàn)spring task定時任務(wù)的配置
因工作需要,需要定時執(zhí)行一個方法,通過相關(guān)比較后,發(fā)現(xiàn)spring自帶的task 可以滿足,配置簡單
步驟
1)增加配置文件 ,在applicationContext-cfg.xml 主配置文件里面添加 相關(guān)task標(biāo)簽
<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類和執(zhí)行方法
編寫jobService類,里面實(shí)現(xiàn)testjobThread方法,調(diào)用的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:開票定時任務(wù)
*/
public void testjobThread()
{
Date startdate = new Date();
logger.info("DZFP_job_JOB 開始執(zhí)行任務(wù)...,時間 " + 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 任務(wù)完成...時間 " + enddate + " 耗時 " + String.valueOf(enddate.getTime() - startdate.getTime()) + "毫秒");
}
3)配置task相關(guān)配置文件,在文件applicationContext-cfg.xml 中增加下列內(nèi)容
pool-size="5" 該參數(shù)主要解決,多個調(diào)度并行的問題,如下圖5個task任務(wù),建議設(shè)置3--5個調(diào)度
如果配置參數(shù)為 1,下面5個task任務(wù)會依次執(zhí)行,如果一個時間超出,后面的任務(wù)一直在等待,影響業(yè)務(wù)
<!-- 定時任務(wù) -->
<task:scheduler id="scheduler" pool-size="5" />
<task:scheduled-tasks scheduler="scheduler">
<!-- 每天7點(diǎn)到7點(diǎn)55, 每隔5分鐘執(zhí)行一次 "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>
使用以上配置后,啟動項(xiàng)目就可以定時執(zhí)行testjobThread方法里面的業(yè)務(wù)了。
2、task里面的一個job方法如何使用多線程,配置線程池
經(jīng)過測試,spring task里面的方法是被串行執(zhí)行的,比如上面配置的方法 testjobThread方法,5秒執(zhí)行一次,如果有一個執(zhí)行過程時間過長,后面的一次調(diào)度一直等上次執(zhí)行結(jié)束后,才會啟動下一次調(diào)用。
也就是說spring task是會監(jiān)控 執(zhí)行方法的主線程,如果主線程未結(jié)束的話,下一次就不會執(zhí)行。
根據(jù)業(yè)務(wù)需求,這個testjobThread里面的 業(yè)務(wù),需要多線程執(zhí)行 (批量抽取數(shù)據(jù))
spring框架里面,推薦使用線程池
1)配置線程池
在applicationContext-cfg.xml文件中增加配置如下
<!-- spring線程池-->
<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<!-- 線程池維護(hù)線程的最少數(shù)量 -->
<property name="corePoolSize" value="5" />
<!-- 線程池維護(hù)線程所允許的空閑時間,默認(rèn)為60s -->
<property name="keepAliveSeconds" value="200" />
<!-- 線程池維護(hù)線程的最大數(shù)量 -->
<property name="maxPoolSize" value="20" />
<!-- 緩存隊(duì)列最大長度 -->
<property name="queueCapacity" value="20" />
<!-- 對拒絕task的處理策略 線程池對拒絕任務(wù)(無線程可用)的處理策略,目前只支持AbortPolicy、CallerRunsPolicy;默認(rèn)為后者-->
<property name="rejectedExecutionHandler">
<!-- AbortPolicy:直接拋出java.util.concurrent.RejectedExecutionException異常 -->
<!-- CallerRunsPolicy:主線程直接執(zhí)行該任務(wù),執(zhí)行完之后嘗試添加下一個任務(wù)到線程池中,可以有效降低向線程池內(nèi)添加任務(wù)的速度 -->
<!-- DiscardOldestPolicy:拋棄舊的任務(wù)、暫不支持;會導(dǎo)致被丟棄的任務(wù)無法再次被執(zhí)行 -->
<!-- DiscardPolicy:拋棄當(dāng)前任務(wù)、暫不支持;會導(dǎo)致被丟棄的任務(wù)無法再次被執(zhí)行 -->
<bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" />
</property>
<property name="waitForTasksToCompleteOnShutdown" value="true" />
</bean>
2)修改業(yè)務(wù)操作類為thread類,實(shí)現(xiàn)run()方法
添加計(jì)數(shù)器CountDownLatch ,控制子線程結(jié)束后,再結(jié)束主線程
注意對象實(shí)現(xiàn)@Scope("prototype"),用到了成員變量參數(shù)
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默認(rèn)是單線程 串行執(zhí)行,即一個方法執(zhí)行完成前,后面的job不會執(zhí)行的
* 但是如果主方法里面產(chǎn)生了thread線程, 主線程如果不等子線程結(jié)束后 就結(jié)束的話, task任務(wù)會產(chǎn)生多次調(diào)度
*/
private String Treadname;
private CountDownLatch latch;
public testThreadAction(String Treadname,CountDownLatch latch){
this.Treadname=Treadname;
this.latch=latch;
}
@Override
public void run()
{
try
{
//主業(yè)務(wù)方法
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{
//設(shè)置實(shí)例 執(zhí)行完畢
latch.countDown();
}
}
public void setTreadname(String treadname)
{
Treadname = treadname;
}
public void setLatch(CountDownLatch latch)
{
this.latch = latch;
}
}
2)修改job調(diào)度的方法為多線程,配置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工具類,類似與計(jì)數(shù)器,主要實(shí)現(xiàn)子線程未結(jié)束錢,主線程一直等待
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(); //子線程未結(jié)束前,一直等待
//test1.run();
}
catch (Exception e)
{
e.printStackTrace();
logger.error(StringUtil.grabExceptionMessage(e));
}
}
}
執(zhí)行效果如下:
雖然 testjobThread 5秒執(zhí)行一次,但是因?yàn)槭褂玫搅?latch.await() latch.countDown();需要等子線程執(zhí)行完畢,才會進(jìn)行下一次job
子線程每次循環(huán),會sleep 20秒,從下面結(jié)果看,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 線程池配置
默認(rèn)線程池ThreadPoolTaskExecutor配置
配置核心參數(shù)
直接在application.properties中配置核心參數(shù)
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-
創(chuàng)建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();
//配置核心線程數(shù)
executor.setCorePoolSize(5);
//配置最大線程數(shù)
executor.setMaxPoolSize(6);
//配置隊(duì)列大小
executor.setQueueCapacity(99999);
//配置線程池中的線程的名稱前綴
executor.setThreadNamePrefix("swy-task-");
// rejection-policy:當(dāng)pool已經(jīng)達(dá)到max size的時候,如何處理新任務(wù)
// CALLER_RUNS:不在新線程中執(zhí)行任務(wù),而是有調(diào)用者所在的線程來執(zhí)行
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
//執(zhí)行初始化
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創(chuàng)建新線程池類
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);
}
}
配置新建線程池類的核心參數(shù)
@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();
//配置核心線程數(shù)
executor.setCorePoolSize(5);
//配置最大線程數(shù)
executor.setMaxPoolSize(8);
//配置隊(duì)列大小
executor.setQueueCapacity(99999);
//配置線程池中的線程的名稱前綴
executor.setThreadNamePrefix("async-service-");
// rejection-policy:當(dāng)pool已經(jīng)達(dá)到max size的時候,如何處理新任務(wù)
// CALLER_RUNS:不在新線程中執(zhí)行任務(wù),而是有調(diào)用者所在的線程來執(zhí)行
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
//執(zhí)行初始化
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");
}
以上為個人經(jīng)驗(yàn),希望能給大家一個參考,也希望大家多多支持腳本之家。
相關(guān)文章
java觀察者模式實(shí)現(xiàn)和java觀察者模式演化
觀察者模式是經(jīng)典設(shè)計(jì)模式中很常用的一種,平常我們看到的監(jiān)聽器,基本上都是采用這種設(shè)計(jì)模式實(shí)現(xiàn)的,這里探討一下觀察者模式的演化2014-02-02
springboot項(xiàng)目實(shí)現(xiàn)斷點(diǎn)續(xù)傳功能
這篇文章主要介紹了springboot項(xiàng)目實(shí)現(xiàn)斷點(diǎn)續(xù)傳,本文通過示例代碼給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2023-08-08
解決Android Studio安裝后運(yùn)行出錯dose not...和Internal error...
這篇文章主要介紹了解決Android Studio安裝后運(yùn)行出錯dose not...和Internal error...的相關(guān)資料,需要的朋友可以參考下2017-03-03
JDBC用IDEA連接SQLServer數(shù)據(jù)庫的超實(shí)用教程
JDBC是Java連接數(shù)據(jù)庫的一種接口,它由各個數(shù)據(jù)庫廠商為開發(fā)者提供的接口,要使用它需要到相應(yīng)廠商下載對應(yīng)的jar包,下面這篇文章主要給大家介紹了關(guān)于JDBC用IDEA連接SQLServer數(shù)據(jù)庫的超實(shí)用教程,需要的朋友可以參考下2023-05-05
SpringBoot讀取資源目錄中JSON文件的方法實(shí)例
最近做項(xiàng)目遇到需要將json類型的配置文件引用到項(xiàng)目中,已經(jīng)將讀取json文件的方法封裝成工具類,下面這篇文章主要給大家介紹了關(guān)于SpringBoot讀取資源目錄中JSON文件的相關(guān)資料,需要的朋友可以參考下2023-04-04
Java類的繼承實(shí)例詳解(動力節(jié)點(diǎn)Java學(xué)院整理)
在Java開發(fā)中,我們常常用到繼承這一概念,可以說繼承是Java這類面向?qū)ο缶幊陶Z言的基石,今天小編一起和大家一起學(xué)習(xí)java類的繼承2017-04-04
Java編程實(shí)現(xiàn)多線程TCP服務(wù)器完整實(shí)例
這篇文章主要介紹了Java編程實(shí)現(xiàn)多線程TCP服務(wù)器完整實(shí)例,具有一定借鑒價(jià)值,需要的朋友可以參考下2018-01-01

