SpringBoot 实现异步调用 | @Async
首先我们来看看在Spring中为什么要使用异步编程,它能解决什么问题?
为什么要用异步框架,它解决什么问题?
在SpringBoot的日常开发中,一般都是同步调用的。但实际中有很多场景非常适合使用异步来处理,如:注册新用户,送100个积分;或下单成功,发送push消息等等。
就拿注册新用户这个用例来说,为什么要异步处理?
- 第一个原因:容错性、健壮性,如果送积分出现异常,不能因为送积分而导致用户注册失败;因为用户注册是主要功能,送积分是次要功能,即使送积分异常也要提示用户注册成功,然后后面在针对积分异常做补偿处理。
- 第二个原因:提升性能,例如注册用户花了20毫秒,送积分花费50毫秒,如果用同步的话,总耗时70毫秒,用异步的话,无需等待积分,故耗时20毫秒。
故,异步能解决2个问题,性能和容错性。
SpringBoot如何实现异步调用?
对于异步方法调用,从Spring3开始提供了
@Async
注解,我们只需要在方法上标注此注解,此方法即可实现异步调用。
当然,我们还需要一个配置类,通过Enable模块驱动注解
@EnableAsync
来开启异步功能。
实现异步调用
第一步:新建配置类,开启@Async功能支持
使用
@EnableAsync
来开启异步任务支持,
@EnableAsync
注解可以直接放在SpringBoot启动类上,也可以单独放在其他配置类上。我们这里选择使用单独的配置类
AsyncConfiguration
。
至于为什么使用线程池,后面会讲到。
importlombok.extern.slf4j.Slf4j;importorg.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.scheduling.annotation.AsyncConfigurer;importorg.springframework.scheduling.annotation.EnableAsync;importorg.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;importjava.util.concurrent.Executor;importjava.util.concurrent.ThreadPoolExecutor;/**
* 描述:异步配置
* 创建人:HuangTuL
*/@Slf4j@Configuration@EnableAsync// 可放在启动类上或单独的配置类publicclassAsyncConfigurationimplementsAsyncConfigurer{@Bean(name ="asyncPoolTaskExecutor")publicThreadPoolTaskExecutorexecutor(){ThreadPoolTaskExecutor taskExecutor =newThreadPoolTaskExecutor();//核心线程数
taskExecutor.setCorePoolSize(10);//线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程
taskExecutor.setMaxPoolSize(100);//缓存队列
taskExecutor.setQueueCapacity(50);//设置线程的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
taskExecutor.setKeepAliveSeconds(200);//异步方法内部线程名称
taskExecutor.setThreadNamePrefix("async-");/**
* 当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize,如果还有任务到来就会采取任务拒绝策略
* 通常有以下四种策略:
* ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
* ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。
* ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)
* ThreadPoolExecutor.CallerRunsPolicy:重试添加当前的任务,自动重复调用 execute() 方法,直到成功
*/
taskExecutor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());
taskExecutor.initialize();return taskExecutor;}/**
* 指定默认线程池
* The {@link Executor} instance to be used when processing async method invocations.
*/@OverridepublicExecutorgetAsyncExecutor(){returnexecutor();}/**
* The {@link AsyncUncaughtExceptionHandler} instance to be used
* when an exception is thrown during an asynchronous method execution
* with {@code void} return type.
*/@OverridepublicAsyncUncaughtExceptionHandlergetAsyncUncaughtExceptionHandler(){return(ex, method, params)-> log.error("线程池执行任务发送未知错误, 执行方法:{}", method.getName(), ex);}}
第二步:在方法上标记异步调用
增加一个Component类,用来进行业务处理,同时添加
@Async
注解,代表该方法为异步处理。
importlombok.SneakyThrows;importlombok.extern.slf4j.Slf4j;importorg.springframework.scheduling.annotation.Async;importorg.springframework.stereotype.Component;/**
* 描述:异步方法调用
* 创建人:HuangTuL
*/@Slf4j@ComponentpublicclassAsyncTask{@SneakyThrows@Async// 在异步配置类设置了默认线程池则不需要再指定线程池名称// @Async("asyncPoolTaskExecutor")publicvoiddoTask1(){long t1 =System.currentTimeMillis();Thread.sleep(2000);long t2 =System.currentTimeMillis();
log.info("task1方法耗时 {} ms", t2-t1);}@SneakyThrows@Async// @Async("otherPoolTaskExecutor") // 其他线程池的名称publicvoiddoTask2(){long t1 =System.currentTimeMillis();Thread.sleep(3000);long t2 =System.currentTimeMillis();
log.info("task2方法耗时 {} ms", t2-t1);}}
第三步:在Controller中进行异步方法调用
importlombok.extern.slf4j.Slf4j;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.RequestMapping;importorg.springframework.web.bind.annotation.RestController;@Slf4j@RestController@RequestMapping("/async")publicclassAsyncController{@AutowiredprivateAsyncTask asyncTask;@RequestMapping("/task")publicvoidtask()throwsInterruptedException{long t1 =System.currentTimeMillis();
asyncTask.doTask1();
asyncTask.doTask2();Thread.sleep(1000);long t2 =System.currentTimeMillis();
log.info("main方法耗时{} ms", t2-t1);}}
通过访问
http://localhost:8080/async/task
查看控制台日志:
[2021-12-01 20:21:36.036] [INFO] [http-nio-8080-exec-1] - task(AsyncController.java:27) - main方法耗时 1009 ms
[2021-12-01 20:21:37.037] [INFO] [async-1] - doTask1(AsyncTask.java:23) - task1方法耗时 2004 ms
[2021-12-01 20:21:38.038] [INFO] [async-2] - doTask2(AsyncTask.java:32) - task2方法耗时 3003 ms
通过日志可以看到:主线程不需要等待异步方法执行完成,减少了响应时间,提高了接口性能。
通过上面三步我们就可以在SpringBoot中使用异步方法来提高我们接口性能了。
为什么要给
@Async
自定义线程池?
使用
@Async
注解,在默认情况下用的是
SimpleAsyncTaskExecutor
线程池,该线程池不是真正意义上的线程池。
使用此线程池无法实现线程重用,每次调用都会新建一条线程。若系统中不断的创建线程,最终会导致系统占用内存过高,引发
OutOfMemoryErro
r错误,关键代码如下:
publicvoidexecute(Runnable task,long startTimeout){Assert.notNull(task,"Runnable must not be null");Runnable taskToUse =this.taskDecorator !=null?this.taskDecorator.decorate(task): task;//判断是否开启限流,默认为否if(this.isThrottleActive()&& startTimeout >0L){//执行前置操作,进行限流this.concurrencyThrottle.beforeAccess();this.doExecute(newSimpleAsyncTaskExecutor.ConcurrencyThrottlingRunnable(taskToUse));}else{//未限流的情况,执行线程任务this.doExecute(taskToUse);}}protectedvoiddoExecute(Runnable task){//不断创建线程Thread thread =this.threadFactory !=null?this.threadFactory.newThread(task):this.createThread(task);
thread.start();}//创建线程publicThreadcreateThread(Runnable runnable){//指定线程名,task-1,task-2...Thread thread =newThread(this.getThreadGroup(), runnable,this.nextThreadName());
thread.setPriority(this.getThreadPriority());
thread.setDaemon(this.isDaemon());return thread;}
我们也可以直接通过上面的控制台日志观察,每次打印的线程名都是[task-1]、[task-2]、[task-3]、[task-4]…递增的。
正因如此,所以我们在使用Spring中的
@Async
异步框架时一定要自定义线程池,替代默认的
SimpleAsyncTaskExecutor
。
Spring提供了多种线程池
SimpleAsyncTaskExecutor
:不是真的线程池,这个类不重用线程,每次调用都会创建一个新的线程。SyncTaskExecutor
:这个类没有实现异步调用,只是一个同步操作。只适用于不需要多线程的地ConcurrentTaskExecutor
:Executor的适配类,不推荐使用。如果ThreadPoolTaskExecutor不满足要求时,才用考虑使用这个类ThreadPoolTaskScheduler
:可以使用cron表达式ThreadPoolTaskExecutor
:最常使用,推荐。其实质是对java.util.concurrent.ThreadPoolExecutor的包装
为
@Async
实现一个自定义线程池
importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.scheduling.annotation.EnableAsync;importorg.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;importjava.util.concurrent.ThreadPoolExecutor;/**
* 描述:异步配置2
* 创建人:HuangTuL
*/@Configuration@EnableAsync// 可放在启动类上或单独的配置类publicclassAsyncConfiguration2{@Bean(name ="asyncPoolTaskExecutor")publicThreadPoolTaskExecutorexecutor(){ThreadPoolTaskExecutor taskExecutor =newThreadPoolTaskExecutor();//核心线程数
taskExecutor.setCorePoolSize(10);//线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程
taskExecutor.setMaxPoolSize(100);//缓存队列
taskExecutor.setQueueCapacity(50);//许的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
taskExecutor.setKeepAliveSeconds(200);//异步方法内部线程名称
taskExecutor.setThreadNamePrefix("async-");/**
* 当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize,如果还有任务到来就会采取任务拒绝策略
* 通常有以下四种策略:
* ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
* ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。
* ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)
* ThreadPoolExecutor.CallerRunsPolicy:重试添加当前的任务,自动重复调用 execute() 方法,直到成功
*/
taskExecutor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());
taskExecutor.initialize();return taskExecutor;}}
配置自定义线程池以后我们就可以大胆的使用
@Async
提供的异步处理能力了。
多个线程池处理
在现实的互联网项目开发中,针对高并发的请求,一般的做法是高并发接口单独线程池隔离处理。
假设现在2个高并发接口:一个是修改用户信息接口,刷新用户redis缓存;一个是下订单接口,发送app push信息。往往会根据接口特征定义两个线程池,这时候我们在使用
@Async
时就需要通过指定线程池名称进行区分。
为
@Async
指定线程池名字
@SneakyThrows@Async("asyncPoolTaskExecutor")publicvoiddoTask1(){long t1 =System.currentTimeMillis();Thread.sleep(2000);long t2 =System.currentTimeMillis();
log.info("task1方法耗时 {} ms", t2-t1);}
当系统存在多个线程池时,我们也可以配置一个默认线程池,对于非默认的异步任务再通过
@Async(“otherTaskExecutor”)
来指定线程池名称。
配置默认线程池
可以修改配置类让其实现
AsyncConfigurer
,并重写
getAsyncExecutor()
方法,指定默认线程池:
importlombok.extern.slf4j.Slf4j;importorg.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.scheduling.annotation.AsyncConfigurer;importorg.springframework.scheduling.annotation.EnableAsync;importorg.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;importjava.util.concurrent.Executor;importjava.util.concurrent.ThreadPoolExecutor;/**
* 描述:异步配置
* 创建人:HuangTuL
*/@Slf4j@Configuration@EnableAsync// 可放在启动类上或单独的配置类publicclassAsyncConfigurationimplementsAsyncConfigurer{@Bean(name ="asyncPoolTaskExecutor")publicThreadPoolTaskExecutorexecutor(){ThreadPoolTaskExecutor taskExecutor =newThreadPoolTaskExecutor();//核心线程数
taskExecutor.setCorePoolSize(10);//线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程
taskExecutor.setMaxPoolSize(100);//缓存队列
taskExecutor.setQueueCapacity(50);//设置线程的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
taskExecutor.setKeepAliveSeconds(200);//异步方法内部线程名称
taskExecutor.setThreadNamePrefix("async-");/**
* 当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize,如果还有任务到来就会采取任务拒绝策略
* 通常有以下四种策略:
* ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
* ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。
* ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)
* ThreadPoolExecutor.CallerRunsPolicy:重试添加当前的任务,自动重复调用 execute() 方法,直到成功
*/
taskExecutor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());
taskExecutor.initialize();return taskExecutor;}/**
* 指定默认线程池
* The {@link Executor} instance to be used when processing async method invocations.
*/@OverridepublicExecutorgetAsyncExecutor(){returnexecutor();}/**
* The {@link AsyncUncaughtExceptionHandler} instance to be used
* when an exception is thrown during an asynchronous method execution
* with {@code void} return type.
*/@OverridepublicAsyncUncaughtExceptionHandlergetAsyncUncaughtExceptionHandler(){return(ex, method, params)-> log.error("线程池执行任务发送未知错误, 执行方法:{}", method.getName(), ex);}}
如下,
doTask1()
方法使用默认使用线程池
asyncPoolTaskExecutor
,
doTask2()
使用线程池
otherTaskExecutor
,非常灵活。
@SneakyThrows@AsyncpublicvoiddoTask1(){long t1 =System.currentTimeMillis();Thread.sleep(2000);long t2 =System.currentTimeMillis();
log.info("task1方法耗时 {} ms", t2-t1);}@SneakyThrows@Async("otherTaskExecutor")publicvoiddoTask2(){long t1 =System.currentTimeMillis();Thread.sleep(3000);long t2 =System.currentTimeMillis();
log.info("task2方法耗时 {} ms", t2-t1);}
使用
@Async
注解可能会导致的问题
如果
serviceA
、
serviceB
对象之间相互依赖,
serviceA
和
serviceB
总一个一个会先实例化,而
serviceA
或
serviceB
里面使用了
@Async
注解,会导致循环依赖异常:
org.springframework.beans.factory.BeanCurrentlyInCreationException
在springboot中,以上报错被捕捉,抛出的异常是:
The dependencies of some of the beans in the application context form a cycle
原因
我们知道,
spring
三级缓存一定程度上解决了循环依赖问题。
A
对象在实例化之后,属性赋值【
opulateBean(beanName, mbd, instanceWrapper)
】执行之前,将
ObjectFactory
添加至三级缓存中,从而使得在
B
对象实例化后的属性赋值过程中,能从三级缓存拿到
ObjectFactory
,调用
getObject()
方法拿到
A
的引用,
B
由此能顺利完成初始化并加入到
IOC
容器。此时A对象完成属性赋值之后,将会执行初始化【
initializeBean(beanName, exposedObject, mbd)
方法】,重点是
@Async
注解的处理正是在这地方完成的,其对应的后置处理器
AsyncAnnotationBeanPostProcessor
,在
postProcessAfterInitialization
方法中将返回代理对象,此代理对象与
B
中持有的A对象引用不同,导致了以上报错。
解决办法
- 1.在A类上加
@Lazy
,保证A对象实例化晚于B对象 - 2.不使用
@Async
注解,通过自定义异步工具类发起异步线程(线程池) - 3.不要让
@Async
的Bean参与循环依赖
@Async
问题解决参考博客
END
版权归原作者 慌途L 所有, 如有侵权,请联系我们删除。