浅谈Spring @Async异步线程池用法总结

2023-09-27 15:12:05 69

本文介绍了Spring@Async异步线程池用法总结，分享给大家，希望对大家有帮助

1.TaskExecutor

spring异步线程池的接口类，其实质是Java.util.concurrent.Executor

Spring已经实现的异常线程池：

1.SimpleAsyncTaskExecutor：不是真的线程池，这个类不重用线程，每次调用都会创建一个新的线程。
2.SyncTaskExecutor：这个类没有实现异步调用，只是一个同步操作。只适用于不需要多线程的地方
3.ConcurrentTaskExecutor：Executor的适配类，不推荐使用。如果ThreadPoolTaskExecutor不满足要求时，才用考虑使用这个类
4.SimpleThreadPoolTaskExecutor：是Quartz的SimpleThreadPool的类。线程池同时被quartz和非quartz使用，才需要使用此类
5.ThreadPoolTaskExecutor：最常使用，推荐。其实质是对java.util.concurrent.ThreadPoolExecutor的包装

2.@Async

spring对过@Async定义异步任务

异步的方法有3种

1.最简单的异步调用，返回值为void
2.带参数的异步调用异步方法可以传入参数
3.异常调用返回Future

详细见代码：

@Component
publicclassAsyncDemo{
privatestaticfinalLoggerlog=LoggerFactory.getLogger(AsyncDemo.class);

/**
*最简单的异步调用，返回值为void
*/
@Async
publicvoidasyncInvokeSimplest(){
log.info("asyncSimplest");
}

/**
*带参数的异步调用异步方法可以传入参数
*
*@params
*/
@Async
publicvoidasyncInvokeWithParameter(Strings){
log.info("asyncInvokeWithParameter,parementer={}",s);
}

/**
*异常调用返回Future
*
*@parami
*@return
*/
@Async
publicFutureasyncInvokeReturnFuture(inti){
log.info("asyncInvokeReturnFuture,parementer={}",i);
Futurefuture;
try{
Thread.sleep(1000*1);
future=newAsyncResult("success:"+i);
}catch(InterruptedExceptione){
future=newAsyncResult("error");
}
returnfuture;
}

}

以上的异步方法和普通的方法调用相同

asyncDemo.asyncInvokeSimplest();
asyncDemo.asyncInvokeWithException("test");
Futurefuture=asyncDemo.asyncInvokeReturnFuture(100);
System.out.println(future.get());

3.Spring开启异步配置

Spring有两种方法启动配置

1.注解
2.XML

3.1通过注解实现

要启动异常方法还需要以下配置

1.@EnableAsync此注解开户异步调用功能

2.publicAsyncTaskExecutortaskExecutor()方法自定义自己的线程池，线程池前缀”Anno-Executor”。如果不定义，则使用系统默认的线程池。

@SpringBootApplication
@EnableAsync//启动异步调用
publicclassAsyncApplicationWithAnnotation{
privatestaticfinalLoggerlog=LoggerFactory.getLogger(AsyncApplicationWithAnnotation.class);

/**
*自定义异步线程池
*@return
*/
@Bean
publicAsyncTaskExecutortaskExecutor(){
ThreadPoolTaskExecutorexecutor=newThreadPoolTaskExecutor();
executor.setThreadNamePrefix("Anno-Executor");
executor.setMaxPoolSize(10);

//设置拒绝策略
executor.setRejectedExecutionHandler(newRejectedExecutionHandler(){
@Override
publicvoidrejectedExecution(Runnabler,ThreadPoolExecutorexecutor){
//.....
}
});
//使用预定义的异常处理类
//executor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());

returnexecutor;
}

publicstaticvoidmain(String[]args){
log.info("StartAsyncApplication..");
SpringApplication.run(AsyncApplicationWithAnnotation.class,args);
}
}

以上的异常方法和普通的方法调用相同

@RunWith(SpringRunner.class)
@SpringBootTest(classes=AsyncApplicationWithAnnotation.class)
publicclassAsyncApplicationWithAnnotationTests{
@Autowired
privateAsyncDemoasyncDemo;

@Test
publicvoidcontextLoads()throwsInterruptedException,ExecutionException{
asyncDemo.asyncInvokeSimplest();
asyncDemo.asyncInvokeWithParameter("test");
Futurefuture=asyncDemo.asyncInvokeReturnFuture(100);
System.out.println(future.get());
}
}

执行测试用例，输出内容如下：

可以看出主线程的名称为main;异步方法则使用Anno-Executor1，可见异常线程池起作用了

2017-03-2820:00:07.731INFO5144---[Anno-Executor1]c.hry.spring.async.annotation.AsyncDemo:asyncSimplest
2017-03-2820:00:07.732INFO5144---[Anno-Executor1]c.hry.spring.async.annotation.AsyncDemo:asyncInvokeWithParameter,parementer=test
2017-03-2820:00:07.751INFO5144---[Anno-Executor1]c.hry.spring.async.annotation.AsyncDemo:asyncInvokeReturnFuture,parementer=100
success:100
2017-03-2820:00:08.757INFO5144---[Thread-2]s.c.a.AnnotationConfigApplicationContext:Closingorg.springframework.context.annotation.AnnotationConfigApplicationContext@47af7f3d:startupdate[TueMar2820:00:06CST2017];rootofcontexthierarchy

3.2通过XML实现

Bean文件配置:spring_async.xml

1.线程的前缀为xmlExecutor
2.启动异步线程池配置

线程池参数说明

1.‘id'：线程的名称的前缀

2.‘pool-size'：线程池的大小。支持范围”min-max”和固定值（此时线程池core和maxsizes相同）

3.‘queue-capacity'：排队队列长度

○Themainideaisthatwhenataskissubmitted,theexecutorwillfirsttrytouseafreethreadifthenumberofactivethreadsiscurrentlylessthanthecoresize.
○Ifthecoresizehasbeenreached,thenthetaskwillbeaddedtothequeueaslongasitscapacityhasnotyetbeenreached.
○Onlythen,ifthequeue'scapacityhasbeenreached,willtheexecutorcreateanewthreadbeyondthecoresize.
○Ifthemaxsizehasalsobeenreached,thentheexecutorwillrejectthetask.
○Bydefault,thequeueisunbounded,butthisisrarelythedesiredconfigurationbecauseitcanleadtoOutOfMemoryErrorsifenoughtasksareaddedtothatqueuewhileallpoolthreadsarebusy.

4.‘rejection-policy':对拒绝的任务处理策略

○InthedefaultThreadPoolExecutor.AbortPolicy,thehandlerthrowsaruntimeRejectedExecutionExceptionuponrejection.
○InThreadPoolExecutor.CallerRunsPolicy,thethreadthatinvokesexecuteitselfrunsthetask.Thisprovidesasimplefeedbackcontrolmechanismthatwillslowdowntheratethatnewtasksaresubmitted.
○InThreadPoolExecutor.DiscardPolicy,ataskthatcannotbeexecutedissimplydropped.
○InThreadPoolExecutor.DiscardOldestPolicy,iftheexecutorisnotshutdown,thetaskattheheadoftheworkqueueisdropped,andthenexecutionisretried(whichcanfailagain,causingthistoberepeated.)

5.‘keep-alive'：线程保活时间（单位秒）

settingdeterminesthetimelimit(inseconds)forwhichthreadsmayremainidlebeforebeingterminated.Iftherearemorethanthecorenumberofthreadscurrentlyinthepool,afterwaitingthisamountoftimewithoutprocessingatask,excessthreadswillgetterminated.Atimevalueofzerowillcauseexcessthreadstoterminateimmediatelyafterexecutingataskwithoutremainingfollow-upworkinthetaskqueue()

异步线程池

@SpringBootApplication
@ImportResource("classpath:/async/spring_async.xml")
publicclassAsyncApplicationWithXML{
privatestaticfinalLoggerlog=LoggerFactory.getLogger(AsyncApplicationWithXML.class);

publicstaticvoidmain(String[]args){
log.info("StartAsyncApplication..");
SpringApplication.run(AsyncApplicationWithXML.class,args);
}
}

测试用例

@RunWith(SpringRunner.class)
@SpringBootTest(classes=AsyncApplicationWithXML.class)
publicclassAsyncApplicationWithXMLTest{
@Autowired
privateAsyncDemoasyncDemo;

@Test
publicvoidcontextLoads()throwsInterruptedException,ExecutionException{
asyncDemo.asyncInvokeSimplest();
asyncDemo.asyncInvokeWithParameter("test");
Futurefuture=asyncDemo.asyncInvokeReturnFuture(100);
System.out.println(future.get());
}
}

运行测试用例，输出内容如下：

可以看出主线程的名称为main;异步方法则使用xmlExecutor-x，可见异常线程池起作用了

2017-03-2820:12:10.540INFO12948---[main]c.h.s.a.xml.AsyncApplicationWithXMLTest:StartedAsyncApplicationWithXMLTestin1.441seconds(JVMrunningfor2.201)
2017-03-2820:12:10.718INFO12948---[xmlExecutor-2]com.hry.spring.async.xml.AsyncDemo:asyncInvokeWithParameter,parementer=test
2017-03-2820:12:10.721INFO12948---[xmlExecutor-1]com.hry.spring.async.xml.AsyncDemo:asyncSimplest
2017-03-2820:12:10.722INFO12948---[xmlExecutor-3]com.hry.spring.async.xml.AsyncDemo:asyncInvokeReturnFuture,parementer=100
success:100
2017-03-2820:12:11.729INFO12948---[Thread-2]s.c.a.AnnotationConfigApplicationContext:Closingorg.springframework.context.annotation.AnnotationConfigApplicationContext@71809907:startupdate[TueMar2820:12:09CST2017];rootofcontexthierarchy

4.对异步方法的异常处理

在调用方法时，可能出现方法中抛出异常的情况。在异步中主要有有两种异常处理方法：

1.对于方法返回值是Futrue的异步方法:a)一种是在调用future的get时捕获异常;b)在异常方法中直接捕获异常

2.对于返回值是void的异步方法：通过AsyncUncaughtExceptionHandler处理异常

AsyncExceptionDemo：

@Component
publicclassAsyncExceptionDemo{
privatestaticfinalLoggerlog=LoggerFactory.getLogger(AsyncExceptionDemo.class);

/**
*最简单的异步调用，返回值为void
*/
@Async
publicvoidasyncInvokeSimplest(){
log.info("asyncSimplest");
}

/**
*带参数的异步调用异步方法可以传入参数
*对于返回值是void，异常会被AsyncUncaughtExceptionHandler处理掉
*@params
*/
@Async
publicvoidasyncInvokeWithException(Strings){
log.info("asyncInvokeWithParameter,parementer={}",s);
thrownewIllegalArgumentException(s);
}

/**
*异常调用返回Future
*对于返回值是Future，不会被AsyncUncaughtExceptionHandler处理，需要我们在方法中捕获异常并处理
*或者在调用方在调用Futrue.get时捕获异常进行处理
*
*@parami
*@return
*/
@Async
publicFutureasyncInvokeReturnFuture(inti){
log.info("asyncInvokeReturnFuture,parementer={}",i);
Futurefuture;
try{
Thread.sleep(1000*1);
future=newAsyncResult("success:"+i);
thrownewIllegalArgumentException("a");
}catch(InterruptedExceptione){
future=newAsyncResult("error");
}catch(IllegalArgumentExceptione){
future=newAsyncResult("error-IllegalArgumentException");
}
returnfuture;
}

}

实现AsyncConfigurer接口对异常线程池更加细粒度的控制

a)创建线程自己的线程池

b)对void方法抛出的异常处理的类AsyncUncaughtExceptionHandler

/**
*通过实现AsyncConfigurer自定义异常线程池，包含异常处理
*
*@authorhry
*
*/
@Service
publicclassMyAsyncConfigurerimplementsAsyncConfigurer{
privatestaticfinalLoggerlog=LoggerFactory.getLogger(MyAsyncConfigurer.class);

@Override
publicExecutorgetAsyncExecutor(){
ThreadPoolTaskExecutorthreadPool=newThreadPoolTaskExecutor();
threadPool.setCorePoolSize(1);
threadPool.setMaxPoolSize(1);
threadPool.setWaitForTasksToCompleteOnShutdown(true);
threadPool.setAwaitTerminationSeconds(60*15);
threadPool.setThreadNamePrefix("MyAsync-");
threadPool.initialize();
returnthreadPool;
}

@Override
publicAsyncUncaughtExceptionHandlergetAsyncUncaughtExceptionHandler(){
returnnewMyAsyncExceptionHandler();
}

/**
*自定义异常处理类
*@authorhry
*
*/
classMyAsyncExceptionHandlerimplementsAsyncUncaughtExceptionHandler{

@Override
publicvoidhandleUncaughtException(Throwablethrowable,Methodmethod,Object...obj){
log.info("Exceptionmessage-"+throwable.getMessage());
log.info("Methodname-"+method.getName());
for(Objectparam:obj){
log.info("Parametervalue-"+param);
}
}

}

}

@SpringBootApplication
@EnableAsync//启动异步调用
publicclassAsyncApplicationWithAsyncConfigurer{
privatestaticfinalLoggerlog=LoggerFactory.getLogger(AsyncApplicationWithAsyncConfigurer.class);

publicstaticvoidmain(String[]args){
log.info("StartAsyncApplication..");
SpringApplication.run(AsyncApplicationWithAsyncConfigurer.class,args);
}


}

测试代码

@RunWith(SpringRunner.class)
@SpringBootTest(classes=AsyncApplicationWithAsyncConfigurer.class)
publicclassAsyncApplicationWithAsyncConfigurerTests{
@Autowired
privateAsyncExceptionDemoasyncDemo;

@Test
publicvoidcontextLoads()throwsInterruptedException,ExecutionException{
asyncDemo.asyncInvokeSimplest();
asyncDemo.asyncInvokeWithException("test");
Futurefuture=asyncDemo.asyncInvokeReturnFuture(100);
System.out.println(future.get());
}
}

运行测试用例

MyAsyncConfigurer捕获AsyncExceptionDemo对象在调用asyncInvokeWithException的异常

2017-04-0216:01:45.591INFO11152---[MyAsync-1]c.h.s.a.exception.AsyncExceptionDemo:asyncSimplest
2017-04-0216:01:45.605INFO11152---[MyAsync-1]c.h.s.a.exception.AsyncExceptionDemo:asyncInvokeWithParameter,parementer=test
2017-04-0216:01:45.608INFO11152---[MyAsync-1]c.h.s.async.exception.MyAsyncConfigurer:Exceptionmessage-test
2017-04-0216:01:45.608INFO11152---[MyAsync-1]c.h.s.async.exception.MyAsyncConfigurer:Methodname-asyncInvokeWithException
2017-04-0216:01:45.608INFO11152---[MyAsync-1]c.h.s.async.exception.MyAsyncConfigurer:Parametervalue-test
2017-04-0216:01:45.608INFO11152---[MyAsync-1]c.h.s.a.exception.AsyncExceptionDemo:asyncInvokeReturnFuture,parementer=100
error-IllegalArgumentException
2017-04-0216:01:46.656INFO11152---[Thread-2]s.c.a.AnnotationConfigApplicationContext:Closingorg.springframework.context.annotation.AnnotationConfigApplicationContext@47af7f3d:startupdate[SunApr0216:01:44CST2017];rootofcontexthierarchy

5.源码地址

代码的GITHUB地址

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

浅谈Spring @Async异步线程池用法总结

热门推荐

随机推荐