您好，登录后才能下订单哦！
密码登录
登录注册
点击 登录注册 即表示同意《亿速云用户服务条款》
小编给大家分享一下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方法里面的业务了。
经过测试，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
配置核心参数
直接在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创建新线程池类
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");
}以上是“spring task和线程池的示例分析”这篇文章的所有内容，感谢各位的阅读！相信大家都有了一定的了解，希望分享的内容对大家有所帮助，如果还想学习更多知识，欢迎关注亿速云行业资讯频道！
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。