文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

spring task和线程池的示例分析

2023-06-22 06:58

关注

小编给大家分享一下spring task和线程池的示例分析,相信大部分人都还不怎么了解,因此分享这篇文章给大家参考一下,希望大家阅读完这篇文章后大有收获,下面让我们一起去了解一下吧!

spring task和线程池的研究

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);          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{      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=8spring.task.execution.pool.max-size=12spring.task.execution.pool.keep-alive=60sspring.task.execution.pool.queue-capacity=100000spring.task.execution.pool.allow-core-thread-timeout=truespring.task.execution.thread-name-prefix=swy-task-

创建JavaBean注入

@Configurationpublic 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@EnableAsyncpublic class MultiThreadApplication { public static void main(String[] args) {  SpringApplication.run(MultiThreadApplication.class, args); }}

在Service层或Controller层的类或方法上添加@Async注解

@Asyncpublic 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);    }}

配置新建线程池类的核心参数

@Configurationpublic 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@EnableAsyncpublic 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和线程池的示例分析”这篇文章的所有内容,感谢各位的阅读!相信大家都有了一定的了解,希望分享的内容对大家有所帮助,如果还想学习更多知识,欢迎关注编程网行业资讯频道!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     807人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     351人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     314人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     433人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     221人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯