spring boot使用自定义配置的线程池执行Async异步任务

2023-09-17 20:07:05 314

在前面的博客中，https://www.nhooo.com/article/106718.htm我们使用了springboot的异步操作，当时，我们使用的是默认的线程池，但是，如果我们想根据项目来定制自己的线程池了，下面就来说说，如何定制线程池！

一、增加配置属性类

packagecom.chhliu.springboot.async.configuration;
importorg.springframework.boot.context.properties.ConfigurationProperties;

@ConfigurationProperties(prefix="spring.task.pool")//该注解的locations已经被启用，现在只要是在环境中，都会优先加载
publicclassTaskThreadPoolConfig{
privateintcorePoolSize;

privateintmaxPoolSize;

privateintkeepAliveSeconds;

privateintqueueCapacity;

…………省略getter,setter方法…………
}

二、创建线程池

packagecom.chhliu.springboot.async.pool;
importjava.util.concurrent.Executor;
importjava.util.concurrent.ThreadPoolExecutor;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.context.annotation.Bean;
importorg.springframework.context.annotation.Configuration;
importorg.springframework.scheduling.annotation.EnableAsync;
importorg.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

importcom.chhliu.springboot.async.configuration.TaskThreadPoolConfig;

@Configuration
@EnableAsync
publicclassTaskExecutePool{

@Autowired
privateTaskThreadPoolConfigconfig;

@Bean
publicExecutormyTaskAsyncPool(){
ThreadPoolTaskExecutorexecutor=newThreadPoolTaskExecutor();
executor.setCorePoolSize(config.getCorePoolSize());
executor.setMaxPoolSize(config.getMaxPoolSize());
executor.setQueueCapacity(config.getQueueCapacity());
executor.setKeepAliveSeconds(config.getKeepAliveSeconds());
executor.setThreadNamePrefix("MyExecutor-");

//rejection-policy：当pool已经达到maxsize的时候，如何处理新任务
//CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
executor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
returnexecutor;
}
}

三、在主类中开启配置支持

packagecom.chhliu.springboot.async;
importorg.springframework.boot.SpringApplication;
importorg.springframework.boot.autoconfigure.SpringBootApplication;
importorg.springframework.boot.context.properties.EnableConfigurationProperties;
importorg.springframework.scheduling.annotation.EnableAsync;

importcom.chhliu.springboot.async.configuration.TaskThreadPoolConfig;

@SpringBootApplication
@EnableAsync
@EnableConfigurationProperties({TaskThreadPoolConfig.class})//开启配置属性支持
publicclassSpringbootAsyncApplication{

publicstaticvoidmain(String[]args){
SpringApplication.run(SpringbootAsyncApplication.class,args);
}
}

四、测试类

packagecom.chhliu.springboot.async.pool;

importorg.slf4j.Logger;
importorg.slf4j.LoggerFactory;
importorg.springframework.scheduling.annotation.Async;
importorg.springframework.stereotype.Component;

@Component
publicclassAsyncTask{
protectedfinalLoggerlogger=LoggerFactory.getLogger(this.getClass());

@Async("myTaskAsyncPool")//myTaskAsynPool即配置线程池的方法名，此处如果不写自定义线程池的方法名，会使用默认的线程池
publicvoiddoTask1(inti)throwsInterruptedException{
logger.info("Task"+i+"started.");
}
}

五、测试

packagecom.chhliu.springboot.async;
importjava.util.concurrent.ExecutionException;
importorg.junit.Test;
importorg.junit.runner.RunWith;
importorg.slf4j.Logger;
importorg.slf4j.LoggerFactory;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.boot.test.context.SpringBootTest;
importorg.springframework.test.context.junit4.SpringRunner;

importcom.chhliu.springboot.async.pool.AsyncTask;

@RunWith(SpringRunner.class)
@SpringBootTest
publicclassSpringbootAsyncApplicationTests{
protectedfinalLoggerlogger=LoggerFactory.getLogger(this.getClass());
@Autowired
privateAsyncTaskasyncTask;

@Test
publicvoidAsyncTaskTest()throwsInterruptedException,ExecutionException{

for(inti=0;i<100;i++){
asyncTask.doTask1(i);
}

logger.info("Alltasksfinished.");
}
}

测试结果如下：

2017-03-2020:15:15.208INFO4068---[MyExecutor-10]c.c.springboot.async.pool.AsyncTask:Task60started.
2017-03-2020:15:15.208INFO4068---[MyExecutor-25]c.c.springboot.async.pool.AsyncTask:Task61started.
2017-03-2020:15:15.208INFO4068---[MyExecutor-6]c.c.springboot.async.pool.AsyncTask:Task62started.
2017-03-2020:15:15.208INFO4068---[MyExecutor-23]c.c.springboot.async.pool.AsyncTask:Task63started.
2017-03-2020:15:15.208INFO4068---[MyExecutor-20]c.c.springboot.async.pool.AsyncTask:Task64started.
2017-03-2020:15:15.208INFO4068---[MyExecutor-19]c.c.springboot.async.pool.AsyncTask:Task65started.
2017-03-2020:15:15.208INFO4068---[MyExecutor-16]c.c.springboot.async.pool.AsyncTask:Task66started.
2017-03-2020:15:15.208INFO4068---[MyExecutor-15]c.c.springboot.async.pool.AsyncTask:Task67started.
2017-03-2020:15:15.208INFO4068---[MyExecutor-12]c.c.springboot.async.pool.AsyncTask:Task68started.
2017-03-2020:15:15.209INFO4068---[MyExecutor-1]c.c.springboot.async.pool.AsyncTask:Task69started.
2017-03-2020:15:15.209INFO4068---[MyExecutor-11]c.c.springboot.async.pool.AsyncTask:Task81started.
2017-03-2020:15:15.209INFO4068---[MyExecutor-8]c.c.springboot.async.pool.AsyncTask:Task82started.
2017-03-2020:15:15.209INFO4068---[MyExecutor-7]c.c.springboot.async.pool.AsyncTask:Task83started.
2017-03-2020:15:15.209INFO4068---[MyExecutor-4]c.c.springboot.async.pool.AsyncTask:Task84started.
2017-03-2020:15:15.209INFO4068---[MyExecutor-29]c.c.springboot.async.pool.AsyncTask:Task85started.
2017-03-2020:15:15.209INFO4068---[MyExecutor-21]c.c.springboot.async.pool.AsyncTask:Task86started.
2017-03-2020:15:15.209INFO4068---[MyExecutor-17]c.c.springboot.async.pool.AsyncTask:Task88started.

测试结果ok！

六、配置默认的线程池

如果我们想使用默认的线程池，但是只是想修改默认线程池的配置，那怎么做了，此时我们需要实现AsyncConfigurer类，示例代码如下：

importjava.lang.reflect.Method;
importjava.util.concurrent.Executor;
importjava.util.concurrent.ThreadPoolExecutor;
importorg.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.context.annotation.Configuration;
importorg.springframework.scheduling.annotation.AsyncConfigurer;
importorg.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
importcom.chhliu.cq.emailservice.threadconfiguration.TaskThreadPoolConfig;
importlombok.extern.slf4j.Slf4j;

/**
*注意：该线程池被所有的异步任务共享，而不属于某一个异步任务
*描述：配置异步任务的线程池
*@authorchhliu
*创建时间：2017年5月22日上午10:20:56
*@version1.2.0
*/
@Slf4j
@Configuration
publicclassAsyncTaskExecutePoolimplementsAsyncConfigurer{

@Autowired
privateTaskThreadPoolConfigconfig;//配置属性类，见上面的代码

@Override
publicExecutorgetAsyncExecutor(){
ThreadPoolTaskExecutorexecutor=newThreadPoolTaskExecutor();
executor.setCorePoolSize(config.getCorePoolSize());
executor.setMaxPoolSize(config.getMaxPoolSize());
executor.setQueueCapacity(config.getQueueCapacity());
executor.setKeepAliveSeconds(config.getKeepAliveSeconds());
executor.setThreadNamePrefix("taskExecutor-");

//rejection-policy：当pool已经达到maxsize的时候，如何处理新任务
//CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
executor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
returnexecutor;
}

@Override
publicAsyncUncaughtExceptionHandlergetAsyncUncaughtExceptionHandler(){//异步任务中异常处理
returnnewAsyncUncaughtExceptionHandler(){

@Override
publicvoidhandleUncaughtException(Throwablearg0,Methodarg1,Object...arg2){
log.error("=========================="+arg0.getMessage()+"=======================",arg0);
log.error("exceptionmethod:"+arg1.getName());
}
};
}
}

使用的时候，只需在方法上加上@Async即可。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。