spring boot使用@Async異步注解的實現(xiàn)原理+源碼
1、java的大部分接口的方法都是串行執(zhí)行的,但是有些業(yè)務(wù)場景是不需要同步返回結(jié)果的,可以把結(jié)果直接返回,具體業(yè)務(wù)異步執(zhí)行,也有些業(yè)務(wù)接口是需要并行獲取數(shù)據(jù),最后把數(shù)據(jù)聚合在統(tǒng)一返回給前端。
通常我們都是采用多線程的方式來實現(xiàn)上述業(yè)務(wù)功能,但spring 提供更優(yōu)雅的方式來實現(xiàn)上述功能,就是@Async 異步注解,在方法上添加@Async,spring就會借助AOP,異步執(zhí)行方法。
1、如何啟用@Async
spring boot通過@EnableAsync 注解啟用@Async異步注解
實現(xiàn)AsyncConfigurer接口,getAsyncExecutor是默認(rèn)自定義的線程池
/** * 線程池配置(@Async) */ @Slf4j @EnableAsync @Configuration public class SimpleExecutorConfig implements AsyncConfigurer { /** 線程池維護線程的最少數(shù)量 */ @Value("${executor.corePoolSize}") private Integer corePoolSize; /** 線程池維護線程的最大數(shù)量 */ @Value("${executor.maxPoolSize}") private Integer maxPoolSize; /** 緩沖隊列的大小 */ @Value("${executor.queueCapacity}") private Integer queueCapacity; /** 為每個線程名設(shè)置一個前綴(1) */ @Value("${executor.threadNamePrefix}") private String threadNamePrefix; /** 為每個線程名設(shè)置一個前綴(2) */ @Value("${executor.threadNamePrefix_2}") private String threadNamePrefix_2; @Bean(ExecutorConstant.simpleExecutor_1) @Override public Executor getAsyncExecutor() { //線程池 ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); taskExecutor.setCorePoolSize(corePoolSize); taskExecutor.setMaxPoolSize(maxPoolSize); taskExecutor.setQueueCapacity(queueCapacity); taskExecutor.setThreadNamePrefix(threadNamePrefix); taskExecutor.initialize(); return taskExecutor; } @Bean(ExecutorConstant.simpleExecutor_2) public Executor asyncExecutor2() { //線程池 ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); taskExecutor.setCorePoolSize(corePoolSize); taskExecutor.setMaxPoolSize(maxPoolSize); taskExecutor.setQueueCapacity(queueCapacity); taskExecutor.setThreadNamePrefix(threadNamePrefix_2); taskExecutor.initialize(); return taskExecutor; } }
2、如何使用@Async
下面是代碼:
TestAsyncService類:
@Slf4j @Service public class TestAsyncService implements ITestAsyncService { /** * 異步方法,無返回值 * @return */ @Async @Override public void asyncFunction_1(){ handleBusinessTime(); log.info("asyncFunction_1 當(dāng)前線程名稱是:{}",Thread.currentThread().getName()); }; /////////////////異步方法,無返回值(指定線程池) start /** * 異步方法,無返回值(指定線程池) * @return */ @Async(value = ExecutorConstant.simpleExecutor_2) @Override public void asyncFunction_2(){ handleBusinessTime(); log.info("asyncFunction_2 當(dāng)前線程名稱是:{}",Thread.currentThread().getName()); }; @Async(ExecutorConstant.simpleExecutor_2) @Override public void asyncFunction_3(){ handleBusinessTime(); log.info("asyncFunction_3 當(dāng)前線程名稱是:{}",Thread.currentThread().getName()); }; /////////////////異步方法,無返回值(指定線程池) end /** * 異步方法,有返回值 * @return */ @Async @Override public Future<Integer> asyncReturnDta_1(){ handleBusinessTime(); log.info("asyncReturnDta_1 當(dāng)前線程名稱是:{}",Thread.currentThread().getName()); return new AsyncResult<Integer>(1); }; /** * 異步方法,有返回值(指定線程池) * @return */ @Async(ExecutorConstant.simpleExecutor_2) @Override public Future<Integer> asyncReturnDta_2(){ handleBusinessTime(); log.info("asyncReturnDta_2 當(dāng)前線程名稱是:{}",Thread.currentThread().getName()); return new AsyncResult<Integer>(1); }; /** * 異步方法,有返回值-超時 * @return */ @Async @Override public Future<Integer> asyncReturnDtaTimeOut(){ handleBusinessTime(); handleBusinessTime(); handleBusinessTime(); handleBusinessTime(); log.info("asyncReturnDta_3 當(dāng)前線程名稱是:{}",Thread.currentThread().getName()); return new AsyncResult<Integer>(1); }; /** * 這方法 模擬處理業(yè)務(wù)或者 去操作數(shù)據(jù)庫 消耗的時間 */ public static void handleBusinessTime(){ //去數(shù)據(jù)庫查詢數(shù)據(jù)耗時 start int[] sleepTime = NumberUtil.generateRandomNumber(2000,5000,1); try { //Thread.sleep 休眠的時候 相當(dāng)于 業(yè)務(wù)操作,或者請求數(shù)據(jù)庫的需要消耗的時間 Thread.sleep(sleepTime[0]); } catch (InterruptedException e) { e.printStackTrace(); } //去數(shù)據(jù)庫查詢數(shù)據(jù)耗時 end } }
TestAsyncController
@Slf4j @RestController @RequestMapping(value = "/v1/async") public class TestAsyncController { @Autowired ITestAsyncService testAsyncService; @ApiOperation(value = "調(diào)用接口") @RequestMapping(value = "/test", method = RequestMethod.GET) public Resp<Integer> test() throws ExecutionException, InterruptedException { log.info("asyncFunction_1 start"); testAsyncService.asyncFunction_1(); log.info("asyncFunction_1 start"); log.info("asyncFunction_2 start"); testAsyncService.asyncFunction_2(); log.info("asyncFunction_2 end"); log.info("asyncFunction_3 start"); testAsyncService.asyncFunction_3(); log.info("asyncFunction_3 end"); log.info("asyncReturnDta_1 & asyncReturnDta_2 start"); Future<Integer> future = testAsyncService.asyncReturnDta_1(); testAsyncService.asyncReturnDta_2(); log.info("asyncReturnDta_1 & asyncReturnDta_2 end"); Integer resp = future.get(); log.info("future.get() resp:{}",resp); return Resp.buildDataSuccess(resp); } @ApiOperation(value = "調(diào)用接口-超時") @RequestMapping(value = "/async_timeOut", method = RequestMethod.GET) public Resp<Integer> async_timeOut() throws ExecutionException, InterruptedException { TimeInterval timeInterval = DateUtil.timer(); log.info("asyncReturnDtaTimeOut start"); Future<Integer> future = testAsyncService.asyncReturnDtaTimeOut(); log.info("asyncReturnDtaTimeOut end"); Integer resp = null; try { //一秒內(nèi) 返回不了數(shù)據(jù)就報錯 resp = future.get(1, TimeUnit.SECONDS); } catch (TimeoutException e) { resp = -1;//請求超時了,相當(dāng)于熔斷,服務(wù)降級 log.error("asyncReturnDtaTimeOut future.get(1, TimeUnit.SECONDS) timeout:",e); } log.info("future.get() resp:{} 耗時:{}毫秒",resp,timeInterval.intervalRestart()); return Resp.buildDataSuccess(resp); } }
/v1/async/test 接口:
2021-06-20 21:09:30.490 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncFunction_1 start
2021-06-20 21:09:30.490 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncFunction_1 start
2021-06-20 21:09:30.490 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncFunction_2 start
2021-06-20 21:09:30.491 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncFunction_2 end
2021-06-20 21:09:30.491 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncFunction_3 start
2021-06-20 21:09:30.491 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncFunction_3 end
2021-06-20 21:09:30.491 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncReturnDta_1 & asyncReturnDta_2 start
2021-06-20 21:09:30.492 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncReturnDta_1 & asyncReturnDta_2 end
2021-06-20 21:09:32.679 INFO 14207 --- [le-1-executor-9] c.e.multi.service.impl.TestAsyncService : asyncFunction_1 當(dāng)前線程名稱是:my-simple-1-executor-9
2021-06-20 21:09:33.454 INFO 14207 --- [le-2-executor-8] c.e.multi.service.impl.TestAsyncService : asyncFunction_3 當(dāng)前線程名稱是:my-simple-2-executor-8
2021-06-20 21:09:33.578 INFO 14207 --- [le-2-executor-9] c.e.multi.service.impl.TestAsyncService : asyncReturnDta_2 當(dāng)前線程名稱是:my-simple-2-executor-9
2021-06-20 21:09:34.101 INFO 14207 --- [e-1-executor-10] c.e.multi.service.impl.TestAsyncService : asyncReturnDta_1 當(dāng)前線程名稱是:my-simple-1-executor-10
2021-06-20 21:09:34.102 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : future.get() resp:1
2021-06-20 21:09:34.357 INFO 14207 --- [le-2-executor-7] c.e.multi.service.impl.TestAsyncService : asyncFunction_2 當(dāng)前線程名稱是:my-simple-2-executor-7
從日志上可以看出,都是異步執(zhí)行的
/v1/async/async_timeOut 接口:
2021-06-20 21:20:58.886 INFO 14427 --- [nio-8666-exec-1] c.e.m.controller.TestAsyncController : asyncReturnDtaTimeOut start
2021-06-20 21:20:58.890 INFO 14427 --- [nio-8666-exec-1] c.e.m.controller.TestAsyncController : asyncReturnDtaTimeOut end
2021-06-20 21:20:59.899 ERROR 14427 --- [nio-8666-exec-1] c.e.m.controller.TestAsyncController : asyncReturnDtaTimeOut future.get(1, TimeUnit.SECONDS) timeout:java.util.concurrent.TimeoutException: null
at java.util.concurrent.FutureTask.get(FutureTask.java:205) ~[na:1.8.0_231]
at com.example.multi.controller.TestAsyncController.async_timeOut(TestAsyncController.java:69) ~[classes/:na]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_231]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_231]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_231]
at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_231]
at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:197) [spring-web-5.3.8.jar:5.3.8]
at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:141) [spring-web-5.3.8.jar:5.3.8]
at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:106) [spring-webmvc-5.3.8.jar:5.3.8]
at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:894) [spring-webmvc-5.3.8.jar:5.3.8]
at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:808) [spring-webmvc-5.3.8.jar:5.3.8]
at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87) [spring-webmvc-5.3.8.jar:5.3.8]
at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1063) [spring-webmvc-5.3.8.jar:5.3.8]
at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:963) [spring-webmvc-5.3.8.jar:5.3.8]
at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1006) [spring-webmvc-5.3.8.jar:5.3.8]
at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:898) [spring-webmvc-5.3.8.jar:5.3.8]
at javax.servlet.http.HttpServlet.service(HttpServlet.java:626) [tomcat-embed-core-9.0.46.jar:4.0.FR]
at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:883) [spring-webmvc-5.3.8.jar:5.3.8]
at javax.servlet.http.HttpServlet.service(HttpServlet.java:733) [tomcat-embed-core-9.0.46.jar:4.0.FR]
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:227) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:53) [tomcat-embed-websocket-9.0.46.jar:9.0.46]
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:100) [spring-web-5.3.8.jar:5.3.8]
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119) [spring-web-5.3.8.jar:5.3.8]
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:93) [spring-web-5.3.8.jar:5.3.8]
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119) [spring-web-5.3.8.jar:5.3.8]
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:201) [spring-web-5.3.8.jar:5.3.8]
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119) [spring-web-5.3.8.jar:5.3.8]
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:202) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:97) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:542) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:143) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:92) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:78) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:357) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:374) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:65) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:893) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1707) [tomcat-embed-core-9.0.46.jar:9.0.46]
at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49) [tomcat-embed-core-9.0.46.jar:9.0.46]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_231]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_231]
at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61) [tomcat-embed-core-9.0.46.jar:9.0.46]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_231]2021-06-20 21:20:59.900 INFO 14427 --- [nio-8666-exec-1] c.e.m.controller.TestAsyncController : future.get() resp:-1 耗時:1014毫秒
2021-06-20 21:21:12.105 INFO 14427 --- [le-1-executor-1] c.e.multi.service.impl.TestAsyncService : asyncReturnDta_3 當(dāng)前線程名稱是:my-simple-1-executor-1
從日志上看出,如果future.get(1, TimeUnit.SECONDS) 到了超時時間,直接拋出超時異常,走主線程后續(xù)代碼。
比較適合 規(guī)定時間范圍內(nèi)要返回數(shù)據(jù)(超時可以根據(jù)業(yè)務(wù)場景,返回一個默認(rèn)值,或者返回值值就是空的)的業(yè)務(wù)場景
@Async原理+源碼
原理:是通過spring aop + 線程池的方式來實現(xiàn)的
源碼:
源碼的方法位置是:AsyncExecutionInterceptor.invoke
107行:是獲取一個線程池108行:如果沒有設(shè)置線程池拋出異常113行:是創(chuàng)建一個線程對象 他的run方法執(zhí)行invocation.proceed()【走實際業(yè)務(wù)代碼】121和124行:走的是統(tǒng)一的異常處理 主要是調(diào)用handleUncaughtException方法,SimpleExecutorConfig 實現(xiàn)了AsyncConfigurer接口它就有g(shù)etAsyncUncaughtExceptionHandler方法,可以重寫這個方法,實現(xiàn)自定義的異常處理
下圖是AsyncConfigurer接口可以實現(xiàn)的方法:
doSubmit方法:
實際上就是調(diào)用線程池的submit方法:
具體代碼,我上傳到gitee,大家感興趣可以clone 傳送門~
@Async源碼參考于:
http://www.dbjr.com.cn/article/141542.htm
以上就是spring boot使用@Async異步注解的實現(xiàn)原理+源碼的詳細(xì)內(nèi)容,更多關(guān)于spring boot異步注解的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
Springboot+Mybatis中typeAliasesPackage正則掃描實現(xiàn)方式
這篇文章主要介紹了Springboot+Mybatis中typeAliasesPackage正則掃描實現(xiàn)方式,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2021-07-07MybatisPlus 不修改全局策略和字段注解如何將字段更新為null
這篇文章主要介紹了MybatisPlus 不修改全局策略和字段注解如何將字段更新為null,本文給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2021-04-04SpringBoot實現(xiàn)無限級評論回復(fù)的項目實踐
本文主要介紹了SpringBoot實現(xiàn)無限級評論回復(fù)的項目實踐,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-03-03SpringMVC 攔截器不攔截靜態(tài)資源的三種處理方式方法
本篇文章主要介紹了SpringMVC 攔截器不攔截靜態(tài)資源的三種處理方式方法,詳細(xì)的介紹了三種方法,有興趣的可以了解一下。2017-01-01