实战Spring Boot 2.0系列(四) - 使用WebAsyncTask处理异步任务

àì夳堔傛蜴生んèń 2022-05-22 12:10 304阅读 0赞

# 前言 #

上文介绍了基于 `@Async` 注解的 **异步调用编程**，本文将继续引入 `Spring Boot` 的 `WebAsyncTask` 进行更灵活异步任务处理，包括 **异步回调**，**超时处理** 和 **异常处理**。

![16413255f98b30e7_w_620_h_595_f_png_s_81732][]

# 正文 #

## 1. 处理线程和异步线程 ##

在开始下面的讲解之前，在这里先区别下两个概念：

1.  处理线程：**处理线程** 属于 `web` 服务器线程，负责 **处理用户请求**，采用 **线程池** 管理。
2.  异步线程：**异步线程** 属于 **用户自定义的线程**，可采用 **线程池管理**。

`Spring` 提供了对 **异步任务** `API`，采用 `WebAsyncTask` 类即可实现 **异步任务**。对异步任务设置相应的 **回调处理**，如当 **任务超时**、**异常抛出** 等。异步任务通常非常实用，比如：当一笔订单支付完成之后，开启异步任务查询订单的支付结果。

## 2. 环境准备 ##

### 配置gradle依赖 ###

利用 `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;
        }
    }

## 3. 正常异步任务 ##

配置一个正常的 `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][http_localhost_8080_completion] ，发起 **正常** 的异步任务请求。

观察控制台输出，可以验证 `WebAsyncTask` 的异步处理流程正常。

请求处理线程：http-nio-8080-exec-2
    继续处理其他事情
    异步工作线程：MvcAsync1
    任务执行完成

`Web` 页面正常响应，页面响应消息如下：

![16413283f62bb949_w_2052_h_152_f_png_s_32271][]

> 注意：WebAsyncTask.onCompletion(Runnable) ：在当前任务执行结束以后，无论是执行成功还是异常中止，onCompletion的回调最终都会被调用。

## 4. 抛出异常异步任务 ##

配置一个 **错误** 的 `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][http_localhost_8080_exception] ，发起 **异常** 的异步任务请求。

`Web` 页面响应异常信息如下：

![16413283fa3d4c0d_w_2050_h_370_f_png_s_80643][]

观察控制台输出，可以验证 `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

## 5. 超时异步任务 ##

配置一个正常的 `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][http_localhost_8080_timeout] ，发起 **超时** 的异步任务请求。

观察控制台输出，可以验证 `WebAsyncTask` 的异步超时处理的过程。

请求处理线程：http-nio-8080-exec-1
    继续处理其他事情
    异步工作线程：MvcAsync3
    任务执行超时
    任务执行完成

`Web` 页面常响应超时提示信息，页面响应消息如下：

![16413283fdc0eed4_w_2052_h_150_f_png_s_25521][]

> 注意：WebAsyncTask.onTimeout(Callable

## 6. 线程池异步任务 ##

上面的三种情况中的 **异步任务** 默认不是采用 **线程池机制** 进行管理的。

也就是说，一个请求进来，虽然释放了处理线程，但是系统依旧会为每个请求创建一个 **异步任务线程**，也就是上面看到的 `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][http_localhost_8080_threadPool] ，观察控制台输出的 **异步线程** 信息，可以发现 **异步任务** 直接从 **线程池** 中获取 **异步线程**。

异步工作线程：asyncTask1
    异步工作线程：asyncTask2
    异步工作线程：asyncTask3
    异步工作线程：asyncTask4
    异步工作线程：asyncTask5
    异步工作线程：asyncTask1
    异步工作线程：asyncTask2
    异步工作线程：asyncTask3
    异步工作线程：asyncTask4
    异步工作线程：asyncTask5

# 小结 #

本文介绍了 `Spring Boot` 提供的 `WebAsyncTask` 的异步编程 `API`。相比上问介绍的 `@Async` 注解，`WebAsyncTask` 提供更加健全的 **超时处理** 和 **异常处理** 支持。

--------------------

欢迎关注技术公众号： 零壹技术栈

![零壹技术栈][1640df3c27eedf87_w_258_h_258_f_jpeg_s_16109]

本帐号将持续分享后端技术干货，包括虚拟机基础，多线程编程，高性能框架，异步、缓存和消息中间件，分布式和微服务，架构学习和进阶等学习资料和文章。

[16413255f98b30e7_w_620_h_595_f_png_s_81732]: https://user-gold-cdn.xitu.io/2018/6/18/16413255f98b30e7?w=620&h=595&f=png&s=81732
[http_localhost_8080_completion]: http://localhost:8080/completion
[16413283f62bb949_w_2052_h_152_f_png_s_32271]: https://user-gold-cdn.xitu.io/2018/6/18/16413283f62bb949?w=2052&h=152&f=png&s=32271
[http_localhost_8080_exception]: http://localhost:8080/exception
[16413283fa3d4c0d_w_2050_h_370_f_png_s_80643]: https://user-gold-cdn.xitu.io/2018/6/18/16413283fa3d4c0d?w=2050&h=370&f=png&s=80643
[http_localhost_8080_timeout]: http://localhost:8080/timeout
[16413283fdc0eed4_w_2052_h_150_f_png_s_25521]: https://user-gold-cdn.xitu.io/2018/6/18/16413283fdc0eed4?w=2052&h=150&f=png&s=25521
[http_localhost_8080_threadPool]: http://localhost:8080/threadPool
[1640df3c27eedf87_w_258_h_258_f_jpeg_s_16109]: https://user-gold-cdn.xitu.io/2018/6/17/1640df3c27eedf87?w=258&h=258&f=jpeg&s=16109