spring?task和線程池的示例分析

小編給大家分享一下spring task和線程池的示例分析，相信大部分人都還不怎么了解，因此分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后大有收獲，下面讓我們一起去了解一下吧！

spring task和線程池的研究

1、如何實(shí)現(xiàn)spring task定時(shí)任務(wù)的配置

因工作需要，需要定時(shí)執(zhí)行一個(gè)方法，通過(guò)相關(guān)比較后，發(fā)現(xiàn)spring自帶的task 可以滿足，配置簡(jiǎn)單

步驟

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注入過(guò)的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:開(kāi)票定時(shí)任務(wù)
    */
    public void testjobThread()
    {
        Date startdate = new Date();
        logger.info("DZFP_job_JOB 開(kāi)始執(zhí)行任務(wù)...,時(shí)間   " + 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ù)完成...時(shí)間  " + enddate + "   耗時(shí)   " + String.valueOf(enddate.getTime() - startdate.getTime()) + "毫秒");
    }

3）配置task相關(guān)配置文件，在文件applicationContext-cfg.xml 中增加下列內(nèi)容

pool-size="5" 該參數(shù)主要解決，多個(gè)調(diào)度并行的問(wèn)題，如下圖5個(gè)task任務(wù)，建議設(shè)置3--5個(gè)調(diào)度

如果配置參數(shù)為 1，下面5個(gè)task任務(wù)會(huì)依次執(zhí)行，如果一個(gè)時(shí)間超出，后面的任務(wù)一直在等待，影響業(yè)務(wù)

 <!-- 定時(shí)任務(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>

使用以上配置后，啟動(dòng)項(xiàng)目就可以定時(shí)執(zhí)行testjobThread方法里面的業(yè)務(wù)了。

2、task里面的一個(gè)job方法如何使用多線程，配置線程池

經(jīng)過(guò)測(cè)試，spring task里面的方法是被串行執(zhí)行的，比如上面配置的方法 testjobThread方法，5秒執(zhí)行一次，如果有一個(gè)執(zhí)行過(guò)程時(shí)間過(guò)長(zhǎng)，后面的一次調(diào)度一直等上次執(zhí)行結(jié)束后，才會(huì)啟動(dòng)下一次調(diào)用。

也就是說(shuō)spring task是會(huì)監(jiān)控 執(zhí)行方法的主線程，如果主線程未結(jié)束的話，下一次就不會(huì)執(zhí)行。

根據(jù)業(yè)務(wù)需求，這個(gè)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ù)線程所允許的空閑時(shí)間，默認(rèn)為60s  -->
        <property name="keepAliveSeconds" value="200" />
        <!-- 線程池維護(hù)線程的最大數(shù)量 -->
        <property name="maxPoolSize" value="20" />
        <!-- 緩存隊(duì)列最大長(zhǎng)度 -->
        <property name="queueCapacity" value="20" />
        <!-- 對(duì)拒絕task的處理策略   線程池對(duì)拒絕任務(wù)（無(wú)線程可用）的處理策略，目前只支持AbortPolicy、CallerRunsPolicy；默認(rèn)為后者-->
        <property name="rejectedExecutionHandler">
        <!-- AbortPolicy:直接拋出java.util.concurrent.RejectedExecutionException異常 -->
            <!-- CallerRunsPolicy:主線程直接執(zhí)行該任務(wù)，執(zhí)行完之后嘗試添加下一個(gè)任務(wù)到線程池中，可以有效降低向線程池內(nèi)添加任務(wù)的速度 -->
            <!-- DiscardOldestPolicy:拋棄舊的任務(wù)、暫不支持；會(huì)導(dǎo)致被丟棄的任務(wù)無(wú)法再次被執(zhí)行 -->
            <!-- DiscardPolicy:拋棄當(dāng)前任務(wù)、暫不支持；會(huì)導(dǎo)致被丟棄的任務(wù)無(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é)束主線程

注意對(duì)象實(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í)行，即一個(gè)方法執(zhí)行完成前，后面的job不會(huì)執(zhí)行的
 * 但是如果主方法里面產(chǎn)生了thread線程， 主線程如果不等子線程結(jié)束后 就結(jié)束的話， task任務(wù)會(huì)產(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("線程號(hào)："+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個(gè)線程

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í)行完畢，才會(huì)進(jìn)行下一次job

子線程每次循環(huán)，會(huì)sleep 20秒，從下面結(jié)果看，3個(gè)線程 每隔20秒才打印一次。符合最終要求

線程號(hào)：29--taskExecutor-3 --test3:---runing--- 0--20170622145500
線程號(hào)：28--taskExecutor-2 --test2:---runing--- 0--20170622145500
線程號(hào)：27--taskExecutor-1 --test1:---runing--- 0--20170622145500
線程號(hào)：28--taskExecutor-2 --test2:---runing--- 1--20170622145520
線程號(hào)：27--taskExecutor-1 --test1:---runing--- 1--20170622145520
線程號(hào)：29--taskExecutor-3 --test3:---runing--- 1--20170622145520
線程號(hào)：29--taskExecutor-3 --test3:---runing--- 2--20170622145540
線程號(hào)：28--taskExecutor-2 --test2:---runing--- 2--20170622145540
線程號(hào)：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的時(shí)候，如何處理新任務(wù)
        // CALLER_RUNS：不在新線程中執(zhí)行任務(wù)，而是有調(diào)用者所在的線程來(lái)執(zhí)行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //執(zhí)行初始化
        executor.initialize();
        return executor;
    }
}

在配置類，或入口類開(kāi)啟@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的時(shí)候，如何處理新任務(wù)
        // CALLER_RUNS：不在新線程中執(zhí)行任務(wù)，而是有調(diào)用者所在的線程來(lái)執(zhí)行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //執(zhí)行初始化
        executor.initialize();
        return executor;
    }
}

在配置類，或入口類開(kāi)啟@EnableAsync注解

@SpringBootApplication
@EnableAsync
public class MultiThreadApplication {
 public static void main(String[] args) {
  SpringApplication.run(MultiThreadApplication.class, args);
 }
}

在Service層或Controller層的類或方法上添加@Async注解，此時(shí)需需注意一定要注明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和線程池的示例分析”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對(duì)大家有所幫助，如果還想學(xué)習(xí)更多知識(shí)，歡迎關(guān)注億速云行業(yè)資訊頻道！