对spring task和线程池的深入研究

网友投稿 212 2022-11-11

目录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标签

xsi:schemaLocation="http://springframework.org/schema/beans

http://springframework.org/schema/beans/spring-beans-3.0.xsd

http://springframework.org/schema/tx

http://springframework.org/schema/tx/spring-tx-3.0.xsd

http://springframework.org/schema/aop

http://springframework.org/schema/aop/spring-aop-3.0.xsd

http://springframework.org/schema/context

http://springframework.org/schema/context/spring-context-3.0.xsd

http://springframework.org/schema/task

http://springframework.org/schema/task/spring-task-3.0.xsd

http://springframework.org/schema/jee

http://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任务会依次执行，如果一个时间超出，后面的任务一直在等待，影响业务

使用以上配置后，启动项目就可以定时执行testjobThread方法里面的业务了。

2、task里面的一个job方法如何使用多线程，配置线程池

经过测试，spring task里面的方法是被串行执行的，比如上面配置的方法 testjobThread方法，5秒执行一次，如果有一个执行过程时间过长，后面的一次调度一直等上次执行结束后，才会启动下一次调用。

也就是说spring task是会监控执行方法的主线程，如果主线程未结束的话，下一次就不会执行。

根据业务需求，这个testjobThread里面的业务，需要多线程执行（批量抽取数据）

spring框架里面，推荐使用线程池

1）配置线程池

在applicationContext-cfg.xml文件中增加配置如下

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.springfhttp://ramework.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 Future submit(Callable 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 ListenableFuture submitListenable(Callable 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")