Spring boot 使用WebAsyncTask处理异步任务

小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

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

正文

1. 处理线程和异步线程

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

处理线程：处理线程属于 web 服务器线程，负责处理用户请求，采用线程池管理。

异步线程：异步线程属于用户自定义的线程，可采用线程池管理。

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<>( * 1000L, () -> {

        out.println(format("异步工作线程：%s", currentThread().getName()));

        // 任务处理时间5s，不超时

        sleep( * 1000L);

        return asyncService.generateUUID();

    });

    // 任务执行完成时调用该方法

    asyncTask.onCompletion(() -> out.println("任务执行完成"));

    out.println("继续处理其他事情");

    return asyncTask;

}

@GetMapping("/completion")

public WebAsyncTask<String> asyncTaskCompletion() {

    // 打印处理线程名

    out.println(format("请求处理线程：%s", currentThread().getName()));

    // 模拟开启一个异步任务，超时时间为10s

    WebAsyncTask<String> asyncTask = new WebAsyncTask<>( * 1000L, () -> {

        out.println(format("异步工作线程：%s", currentThread().getName()));

        // 任务处理时间5s，不超时

        sleep( * 1000L);

        return asyncService.generateUUID();

    });

    // 任务执行完成时调用该方法

    asyncTask.onCompletion(() -> out.println("任务执行完成"));

    out.println("继续处理其他事情");

    return asyncTask;

}

启动 Spring Boot 项目，访问 http://localhost:8080/completion ，发起正常的异步任务请求。

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

请求处理线程：http-nio--exec-

继续处理其他事情

异步工作线程：MvcAsync1

任务执行完成

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

注意：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<>( * 1000L, () -> {

        out.println(format("异步工作线程：%s", currentThread().getName()));

        // 任务处理时间5s，不超时

        sleep( * 1000L);

        throw new Exception(ERROR_MESSAGE);

    });

    // 任务执行完成时调用该方法

    asyncTask.onCompletion(() -> out.println("任务执行完成"));

    asyncTask.onError(() -> {

        out.println("任务执行异常");

        return ERROR_MESSAGE;

    });

    out.println("继续处理其他事情");

    return asyncTask;

}

@GetMapping("/exception")

public WebAsyncTask<String> asyncTaskException() {

    // 打印处理线程名

    out.println(format("请求处理线程：%s", currentThread().getName()));

    // 模拟开启一个异步任务，超时时间为10s

    WebAsyncTask<String> asyncTask = new WebAsyncTask<>( * 1000L, () -> {

        out.println(format("异步工作线程：%s", currentThread().getName()));

        // 任务处理时间5s，不超时

        sleep( * 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--exec-

继续处理其他事情

异步工作线程：MvcAsync2

-- ::10.110 ERROR  --- [nio--exec-] 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$(WebAsyncController.java:) ~[classes/:na]

    at org.springframework.web.context.request.async.WebAsyncManager.lambda$startCallableProcessing$(WebAsyncManager.java:) ~[spring-web-5.0..RELEASE.jar:5.0..RELEASE]

    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:) ~[na:1.8.0_172]

    at java.util.concurrent.FutureTask.run(FutureTask.java:) ~[na:1.8.0_172]

    at java.lang.Thread.run(Thread.java:) [na:1.8.0_172]

-- ::10.111 ERROR  --- [nio--exec-] 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$(WebAsyncController.java:) ~[classes/:na]

    at org.springframework.web.context.request.async.WebAsyncManager.lambda$startCallableProcessing$(WebAsyncManager.java:) ~[spring-web-5.0..RELEASE.jar:5.0..RELEASE]

    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:) ~[na:1.8.0_172]

    at java.util.concurrent.FutureTask.run(FutureTask.java:) ~[na:1.8.0_172]

    at java.lang.Thread.run(Thread.java:) [na:1.8.0_172]

任务执行异常

-- ::10.144  WARN  --- [nio--exec-] 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:) ~[spring-core-5.0..RELEASE.jar:5.0..RELEASE]

    at org.springframework.web.context.request.async.StandardServletAsyncWebRequest.dispatch(StandardServletAsyncWebRequest.java:) ~[spring-web-5.0..RELEASE.jar:5.0..RELEASE]

    at org.springframework.web.context.request.async.WebAsyncManager.setConcurrentResultAndDispatch(WebAsyncManager.java:) ~[spring-web-5.0..RELEASE.jar:5.0..RELEASE]

    at org.springframework.web.context.request.async.WebAsyncManager.lambda$startCallableProcessing$(WebAsyncManager.java:) ~[spring-web-5.0..RELEASE.jar:5.0..RELEASE]

    at org.springframework.web.context.request.async.StandardServletAsyncWebRequest.lambda$onError$(StandardServletAsyncWebRequest.java:) ~[spring-web-5.0..RELEASE.jar:5.0..RELEASE]

    at java.util.ArrayList.forEach(ArrayList.java:) ~[na:1.8.0_172]

    at org.springframework.web.context.request.async.StandardServletAsyncWebRequest.onError(StandardServletAsyncWebRequest.java:) ~[spring-web-5.0..RELEASE.jar:5.0..RELEASE]

    at org.apache.catalina.core.AsyncListenerWrapper.fireOnError(AsyncListenerWrapper.java:) ~[tomcat-embed-core-8.5..jar:8.5.]

    at org.apache.catalina.core.AsyncContextImpl.setErrorState(AsyncContextImpl.java:) ~[tomcat-embed-core-8.5..jar:8.5.]

    at org.apache.catalina.connector.CoyoteAdapter.asyncDispatch(CoyoteAdapter.java:) [tomcat-embed-core-8.5..jar:8.5.]

    at org.apache.coyote.AbstractProcessor.dispatch(AbstractProcessor.java:) [tomcat-embed-core-8.5..jar:8.5.]

    at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:) [tomcat-embed-core-8.5..jar:8.5.]

    at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:) [tomcat-embed-core-8.5..jar:8.5.]

    at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:) [tomcat-embed-core-8.5..jar:8.5.]

    at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:) [tomcat-embed-core-8.5..jar:8.5.]

    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:) [na:1.8.0_172]

    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:) [na:1.8.0_172]

    at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:) [tomcat-embed-core-8.5..jar:8.5.]

    at java.lang.Thread.run(Thread.java:) [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<>( * 1000L, () -> {

        out.println(format("异步工作线程：%s", currentThread().getName()));

        // 任务处理时间5s，不超时

        sleep( * 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--exec-

继续处理其他事情

异步工作线程：MvcAsync3

任务执行超时

任务执行完成

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

注意：WebAsyncTask.onTimeout(Callable

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

也就是说，一个请求进来，虽然释放了处理线程，但是系统依旧会为每个请求创建一个异步任务线程，也就是上面看到的 MvcAsync 开头的异步任务线程。

后果就是开销严重，所以通常采用线程池进行统一的管理，直接在 WebAsyncTask 类构造器传入一个 ThreadPoolTaskExecutor 对象实例即可。

构造一个线程池 Bean 对象：

@Configuration

public class TaskConfiguration {

    @Bean("taskExecutor")

    public ThreadPoolTaskExecutor taskExecutor() {

        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();

        taskExecutor.setCorePoolSize();

        taskExecutor.setMaxPoolSize();

        taskExecutor.setQueueCapacity();

        taskExecutor.setThreadNamePrefix("asyncTask");

        return taskExecutor;

    }

}

在控制器中注入 ThreadPoolTaskExecutor 对象，重新配置基于线程池的异步任务处理。

@Autowired

@Qualifier("taskExecutor")

private ThreadPoolTaskExecutor executor;

@GetMapping("/threadPool")

public WebAsyncTask<String> asyncTaskThreadPool() {

    return new WebAsyncTask<>( * 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