上文介绍了基于 @Async
注解的 异步调用编程 ,本文将继续引入 Spring Boot
的 WebAsyncTask
进行更灵活异步任务处理,包括 异步回调 , 超时处理 和 异常处理 。
在开始下面的讲解之前,在这里先区别下两个概念:
处理线程: 处理线程 属于 web
服务器线程,负责 处理用户请求 ,采用 线程池 管理。
异步线程: 异步线程 属于 用户自定义的线程 ,可采用 线程池管理 。
Spring
提供了对 异步任务 API
,采用 WebAsyncTask
类即可实现 异步任务 。对异步任务设置相应的 回调处理 ,如当 任务超时 、 异常抛出 等。异步任务通常非常实用,比如:当一笔订单支付完成之后,开启异步任务查询订单的支付结果。
利用 Spring Initializer
创建一个 gradle
项目 spring-boot-web-async-task
,创建时添加相关依赖。得到的初始 build.gradle
如下:
buildscript { ext { springBootVersion = '2.0.3.RELEASE' } repositories { mavenCentral() } dependencies { classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}") } } apply plugin: 'java' apply plugin: 'eclipse' apply plugin: 'org.springframework.boot' apply plugin: 'io.spring.dependency-management' group = 'io.ostenant.springboot.sample' version = '0.0.1-SNAPSHOT' sourceCompatibility = 1.8 repositories { mavenCentral() } dependencies { compile('org.springframework.boot:spring-boot-starter-web') testCompile('org.springframework.boot:spring-boot-starter-test') }
配置一个用于异步任务调度的 Mock
服务。
@Service public class WebAsyncService { public String generateUUID() { return UUID.randomUUID().toString(); } }
配置异步处理控制器并注入以上服务 Bean
。
@RestController public class WebAsyncController { private final WebAsyncService asyncService; private final static String ERROR_MESSAGE = "Task error"; private final static String TIME_MESSAGE = "Task timeout"; @Autowired public WebAsyncController(WebAsyncService asyncService) { this.asyncService = asyncService; } }
配置一个正常的 WebAsyncTask
任务对象,设置任务 超时时间 为 10s
。异步任务执行采用 Thread.sleep(long)
模拟,这里设置 异步线程 睡眠时间为 5s
。
@GetMapping("/completion") public WebAsyncTask<String> asyncTaskCompletion() { // 打印处理线程名 out.println(format("请求处理线程:%s", currentThread().getName())); // 模拟开启一个异步任务,超时时间为10s WebAsyncTask<String> asyncTask = new WebAsyncTask<>(10 * 1000L, () -> { out.println(format("异步工作线程:%s", currentThread().getName())); // 任务处理时间5s,不超时 sleep(5 * 1000L); return asyncService.generateUUID(); }); // 任务执行完成时调用该方法 asyncTask.onCompletion(() -> out.println("任务执行完成")); out.println("继续处理其他事情"); return asyncTask; }
启动 Spring Boot
项目,访问 http://localhost:8080/completion ,发起 正常 的异步任务请求。
观察控制台输出,可以验证 WebAsyncTask
的异步处理流程正常。
请求处理线程:http-nio-8080-exec-2 继续处理其他事情 异步工作线程:MvcAsync1 任务执行完成
Web
页面正常响应,页面响应消息如下:
注意:WebAsyncTask.onCompletion(Runnable) :在当前任务执行结束以后,无论是执行成功还是异常中止,onCompletion的回调最终都会被调用。
配置一个 错误 的 WebAsyncTask
任务对象,设置任务 超时时间 为 10s
。在异步任务执行方法中 抛出异常 。
@GetMapping("/exception") public WebAsyncTask<String> asyncTaskException() { // 打印处理线程名 out.println(format("请求处理线程:%s", currentThread().getName())); // 模拟开启一个异步任务,超时时间为10s WebAsyncTask<String> asyncTask = new WebAsyncTask<>(10 * 1000L, () -> { out.println(format("异步工作线程:%s", currentThread().getName())); // 任务处理时间5s,不超时 sleep(5 * 1000L); throw new Exception(ERROR_MESSAGE); }); // 任务执行完成时调用该方法 asyncTask.onCompletion(() -> out.println("任务执行完成")); asyncTask.onError(() -> { out.println("任务执行异常"); return ERROR_MESSAGE; }); out.println("继续处理其他事情"); return asyncTask; }
启动 Spring Boot
项目,访问 http://localhost:8080/exception ,发起 异常 的异步任务请求。
Web
页面响应异常信息如下:
观察控制台输出,可以验证 WebAsyncTask
对于 异常请求 的异步处理过程。
请求处理线程:http-nio-8080-exec-1 继续处理其他事情 异步工作线程:MvcAsync2 2018-06-18 21:12:10.110 ERROR 89875 --- [nio-8080-exec-2] o.a.c.c.C.[.[.[/].[dispatcherServlet] : Servlet.service() for servlet [dispatcherServlet] threw exception java.lang.Exception: Task error at io.ostenant.springboot.sample.controller.WebAsyncController.lambda$asyncTaskException$2(WebAsyncController.java:55) ~[classes/:na] at org.springframework.web.context.request.async.WebAsyncManager.lambda$startCallableProcessing$4(WebAsyncManager.java:317) ~[spring-web-5.0.7.RELEASE.jar:5.0.7.RELEASE] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_172] at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_172] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_172] 2018-06-18 21:12:10.111 ERROR 89875 --- [nio-8080-exec-2] o.a.c.c.C.[.[.[/].[dispatcherServlet] : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is java.lang.Exception: Task error] with root cause java.lang.Exception: Task error at io.ostenant.springboot.sample.controller.WebAsyncController.lambda$asyncTaskException$2(WebAsyncController.java:55) ~[classes/:na] at org.springframework.web.context.request.async.WebAsyncManager.lambda$startCallableProcessing$4(WebAsyncManager.java:317) ~[spring-web-5.0.7.RELEASE.jar:5.0.7.RELEASE] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_172] at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_172] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_172] 任务执行异常 2018-06-18 21:12:10.144 WARN 89875 --- [nio-8080-exec-2] o.apache.catalina.core.AsyncContextImpl : onError() failed for listener of type [org.apache.catalina.core.AsyncListenerWrapper] java.lang.IllegalArgumentException: Cannot dispatch without an AsyncContext at org.springframework.util.Assert.notNull(Assert.java:193) ~[spring-core-5.0.7.RELEASE.jar:5.0.7.RELEASE] at org.springframework.web.context.request.async.StandardServletAsyncWebRequest.dispatch(StandardServletAsyncWebRequest.java:131) ~[spring-web-5.0.7.RELEASE.jar:5.0.7.RELEASE] at org.springframework.web.context.request.async.WebAsyncManager.setConcurrentResultAndDispatch(WebAsyncManager.java:353) ~[spring-web-5.0.7.RELEASE.jar:5.0.7.RELEASE] at org.springframework.web.context.request.async.WebAsyncManager.lambda$startCallableProcessing$2(WebAsyncManager.java:304) ~[spring-web-5.0.7.RELEASE.jar:5.0.7.RELEASE] at org.springframework.web.context.request.async.StandardServletAsyncWebRequest.lambda$onError$0(StandardServletAsyncWebRequest.java:146) ~[spring-web-5.0.7.RELEASE.jar:5.0.7.RELEASE] at java.util.ArrayList.forEach(ArrayList.java:1257) ~[na:1.8.0_172] at org.springframework.web.context.request.async.StandardServletAsyncWebRequest.onError(StandardServletAsyncWebRequest.java:146) ~[spring-web-5.0.7.RELEASE.jar:5.0.7.RELEASE] at org.apache.catalina.core.AsyncListenerWrapper.fireOnError(AsyncListenerWrapper.java:49) ~[tomcat-embed-core-8.5.31.jar:8.5.31] at org.apache.catalina.core.AsyncContextImpl.setErrorState(AsyncContextImpl.java:397) ~[tomcat-embed-core-8.5.31.jar:8.5.31] at org.apache.catalina.connector.CoyoteAdapter.asyncDispatch(CoyoteAdapter.java:239) [tomcat-embed-core-8.5.31.jar:8.5.31] at org.apache.coyote.AbstractProcessor.dispatch(AbstractProcessor.java:232) [tomcat-embed-core-8.5.31.jar:8.5.31] at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:53) [tomcat-embed-core-8.5.31.jar:8.5.31] at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:790) [tomcat-embed-core-8.5.31.jar:8.5.31] at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1468) [tomcat-embed-core-8.5.31.jar:8.5.31] at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49) [tomcat-embed-core-8.5.31.jar:8.5.31] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_172] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_172] at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61) [tomcat-embed-core-8.5.31.jar:8.5.31] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_172] 任务执行完成
注意:WebAsyncTask.onError(Callable<?>) :当异步任务抛出异常的时候,onError()方法即会被调用。
配置一个正常的 WebAsyncTask
任务对象,设置任务 超时时间 为 10s
。异步任务执行采用 Thread.sleep(long)
模拟,这里设置 异步线程 睡眠时间为 15s
,引发异步任务超时。
@GetMapping("/timeout") public WebAsyncTask<String> asyncTaskTimeout() { // 打印处理线程名 out.println(format("请求处理线程:%s", currentThread().getName())); // 模拟开启一个异步任务,超时时间为10s WebAsyncTask<String> asyncTask = new WebAsyncTask<>(10 * 1000L, () -> { out.println(format("异步工作线程:%s", currentThread().getName())); // 任务处理时间5s,不超时 sleep(15 * 1000L); return TIME_MESSAGE; }); // 任务执行完成时调用该方法 asyncTask.onCompletion(() -> out.println("任务执行完成")); asyncTask.onTimeout(() -> { out.println("任务执行超时"); return TIME_MESSAGE; }); out.println("继续处理其他事情"); return asyncTask; }
启动 Spring Boot
项目,访问 http://localhost:8080/timeout ,发起 超时 的异步任务请求。
观察控制台输出,可以验证 WebAsyncTask
的异步超时处理的过程。
请求处理线程:http-nio-8080-exec-1 继续处理其他事情 异步工作线程:MvcAsync3 任务执行超时 任务执行完成
Web
页面常响应超时提示信息,页面响应消息如下:
注意:WebAsyncTask.onTimeout(Callable<?>) :当异步任务发生超时的时候,onTimeout()方法即会被调用。
上面的三种情况中的 异步任务 默认不是采用 线程池机制 进行管理的。
也就是说,一个请求进来,虽然释放了处理线程,但是系统依旧会为每个请求创建一个 异步任务线程 ,也就是上面看到的 MvcAsync
开头的 异步任务线程 。
后果就是开销严重,所以通常采用 线程池 进行统一的管理,直接在 WebAsyncTask
类构造器传入一个 ThreadPoolTaskExecutor
对象实例即可。
构造一个线程池 Bean
对象:
@Configuration public class TaskConfiguration { @Bean("taskExecutor") public ThreadPoolTaskExecutor taskExecutor() { ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); taskExecutor.setCorePoolSize(5); taskExecutor.setMaxPoolSize(10); taskExecutor.setQueueCapacity(10); taskExecutor.setThreadNamePrefix("asyncTask"); return taskExecutor; } }
在控制器中注入 ThreadPoolTaskExecutor
对象,重新配置基于 线程池 的 异步任务处理 。
@Autowired @Qualifier("taskExecutor") private ThreadPoolTaskExecutor executor; @GetMapping("/threadPool") public WebAsyncTask<String> asyncTaskThreadPool() { return new WebAsyncTask<>(10 * 1000L, executor, () -> { out.println(format("异步工作线程:%s", currentThread().getName())); return asyncService.generateUUID(); }); }
并发地请求 http://localhost:8080/threadPool ,观察控制台输出的 异步线程 信息,可以发现 异步任务 直接从 线程池 中获取 异步线程 。
异步工作线程:asyncTask1 异步工作线程:asyncTask2 异步工作线程:asyncTask3 异步工作线程:asyncTask4 异步工作线程:asyncTask5 异步工作线程:asyncTask1 异步工作线程:asyncTask2 异步工作线程:asyncTask3 异步工作线程:asyncTask4 异步工作线程:asyncTask5