實戰Spring Boot 2.0系列(四) - 使用WebAsyncTask處理異步任務

時間  2019-11-08
標籤 實戰 spring boot 2.0 系列 使用 webasynctask 處理 異步 任務 欄目 Spring 简体版
原文   原文鏈接

# 前言

上文介紹了基於 @Async 註解的 異步調用編程，本文將繼續引入 Spring Boot 的 WebAsyncTask 進行更靈活異步任務處理，包括 異步回調，超時處理 和 異常處理。java

本系列文章

  1. 實戰Spring Boot 2.0系列(一) - 使用Gradle構建Docker鏡像
  2. 實戰Spring Boot 2.0系列(二) - 全局異常處理和測試
  3. 實戰Spring Boot 2.0系列(三) - 使用@Async進行異步調用詳解
  4. 實戰Spring Boot 2.0系列(四) - 使用WebAsyncTask處理異步任務
  5. 實戰Spring Boot 2.0系列(五) - Listener, Servlet, Filter和Interceptor
  6. 實戰Spring Boot 2.0系列(六) - 單機定時任務的幾種實現

正文

1. 處理線程和異步線程

在開始下面的講解以前，在這裏先區別下兩個概念：web

  1. 處理線程：處理線程 屬於 web 服務器線程，負責 處理用戶請求，採用 線程池 管理。spring

  2. 異步線程：異步線程 屬於 用戶自定義的線程，可採用 線程池管理。apache

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。tomcat

@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。springboot

@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 ，發起 正常 的異步任務請求。bash

觀察控制檯輸出，能夠驗證 WebAsyncTask 的異步處理流程正常。

請求處理線程：http-nio-8080-exec-2
繼續處理其餘事情
異步工做線程：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<>(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()方法即會被調用。

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 ，發起 超時 的異步任務請求。

觀察控制檯輸出，能夠驗證 WebAsyncTask 的異步超時處理的過程。

請求處理線程：http-nio-8080-exec-1
繼續處理其餘事情
異步工做線程：MvcAsync3
任務執行超時
任務執行完成
複製代碼

Web 頁面常響應超時提示信息，頁面響應消息以下：

注意：WebAsyncTask.onTimeout(Callable<?>) ：當異步任務發生超時的時候，onTimeout()方法即會被調用。

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 ，觀察控制檯輸出的 異步線程 信息，能夠發現 異步任務 直接從 線程池 中獲取 異步線程。

異步工做線程：asyncTask1
異步工做線程：asyncTask2
異步工做線程：asyncTask3
異步工做線程：asyncTask4
異步工做線程：asyncTask5
異步工做線程：asyncTask1
異步工做線程：asyncTask2
異步工做線程：asyncTask3
異步工做線程：asyncTask4
異步工做線程：asyncTask5
複製代碼

小結

本文介紹了 Spring Boot 提供的 WebAsyncTask 的異步編程 API。相比上問介紹的 @Async 註解，WebAsyncTask 提供更加健全的 超時處理 和 異常處理 支持。

歡迎關注技術公衆號： 零壹技術棧

零壹技術棧

本賬號將持續分享後端技術乾貨，包括虛擬機基礎，多線程編程，高性能框架，異步、緩存和消息中間件，分佈式和微服務，架構學習和進階等學習資料和文章。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程