本文在提供完整代碼示例，可見?https://github.com/YunaiV/SpringBoot-Labs?的?lab-29?目錄。

原創不易，給點個?Star?嘿，一起沖鴨！

1. 概述

在日常開發中，我們的邏輯都是同步調用，順序執行。在一些場景下，我們會希望異步調用，將和主線程關聯度低的邏輯異步調用，以實現讓主線程更快的執行完成，提升性能。例如說：記錄用戶訪問日志到數據庫，記錄管理員操作日志到數據庫中。

異步調用，對應的是同步調用。

• 同步調用：指程序按照 定義順序 依次執行，每一行程序都必須等待上一行程序執行完成之后才能執行；

• 異步調用：指程序在順序執行時，不等待異步調用的語句返回結果，就執行后面的程序。

考慮到異步調用的可靠性，我們一般會考慮引入分布式消息隊列，例如說 RabbitMQ、RocketMQ、Kafka 等等。但是在一些時候，我們并不需要這么高的可靠性，可以使用進程內的隊列或者線程池。例如說示例代碼如下：

public class Demo {public static void main(String[] args) {// 創建線程池。這里只是臨時測試，不要扣艿艿遵守阿里 Java 開發規范，YEAHExecutorService executor = Executors.newFixedThreadPool(10);// 提交任務到線程池中執行。executor.submit(new Runnable() {@Overridepublic void run() {System.out.println("聽說我被異步調用了");}});}}

友情提示：這里說進程內的隊列或者線程池，相對不可靠的原因是，隊列和線程池中的任務僅僅存儲在內存中，如果 JVM 進程被異常關閉，將會導致丟失，未被執行。

而分布式消息隊列，異步調用會以一個消息的形式，存儲在消息隊列的服務器上，所以即使 JVM 進程被異常關閉，消息依然在消息隊列的服務器上。

所以，使用進程內的隊列或者線程池來實現異步調用的話，一定要盡可能的保證 JVM 進程的優雅關閉，保證它們在關閉前被執行完成。

在?Spring Framework?的?Spring Task?模塊，提供了?@Async?注解，可以添加在方法上，自動實現該方法的異步調用。

😈 簡單來說，我們可以像使用?@Transactional?聲明式事務，使用 Spring Task 提供的?@Async?注解，😈 聲明式異步。而在實現原理上，也是基于 Spring AOP 攔截，實現異步提交該操作到線程池中，達到異步調用的目的。

如果胖友看過艿艿寫的?《芋道 Spring Boot 定時任務入門》?文章，就會發現 Spring Task 模塊，還提供了定時任務的功能。

下面，讓我們一起遨游 Spring 異步任務的海洋。

2. 快速入門

示例代碼對應倉庫：lab-29-async-demo?。

本小節，我們會編寫示例，對比同步調用和異步調用的性能差別，并演示 Spring?@Async?注解的使用方式。

2.1 引入依賴

在?pom.xml?文件中，引入相關依賴。

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.2.1.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><modelVersion>4.0.0</modelVersion><artifactId>lab-29-async-demo</artifactId><dependencies><!-- 引入 Spring Boot 依賴 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><!-- 方便等會寫單元測試 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies></project>

因為 Spring Task 是 Spring Framework 的模塊，所以在我們引入?spring-boot-web?依賴后，無需特別引入它。

2.2 Application

創建?Application.java?類，配置?@SpringBootApplication?注解。代碼如下：

@SpringBootApplication @EnableAsync // 開啟 @Async 的支持 public class Application {public static void main(String[] args) {SpringApplication.run(Application.class, args);}}

• 在類上添加?@EnableAsync?注解，啟用異步功能。

2.3 DemoService

在?cn.iocoder.springboot.lab29.asynctask.service?包路徑下，創建?DemoService?類。代碼如下：

// DemoService.java@Service public class DemoService {private Logger logger = LoggerFactory.getLogger(getClass());public Integer execute01() {logger.info("[execute01]");sleep(10);return 1;}public Integer execute02() {logger.info("[execute02]");sleep(5);return 2;}private static void sleep(int seconds) {try {Thread.sleep(seconds * 1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}}

• 定義了?#execute01()?和?#execute02()?方法，分別 sleep 10 秒和 5 秒，模擬耗時操作。

• 同時在每個方法里，使用?logger?打印日志，方便我們看到每個方法的開始執行時間，和執行所在線程。

2.4 同步調用測試

創建?DemoServiceTest?測試類，編寫?#task01()?方法，同步調用 DemoService 的上述兩個方法。代碼如下：

// DemoServiceTest.java@RunWith(SpringRunner.class) @SpringBootTest(classes = Application.class) public class DemoServiceTest {private Logger logger = LoggerFactory.getLogger(getClass());@Autowiredprivate DemoService demoService;@Testpublic void task01() {long now = System.currentTimeMillis();logger.info("[task01][開始執行]");demoService.execute01();demoService.execute02();logger.info("[task01][結束執行，消耗時長 {} 毫秒]", System.currentTimeMillis() - now);}}

運行單元測試，執行日志如下：

2019-11-30 14:03:35.820 INFO 64639 --- [ main] c.i.s.l.a.service.DemoServiceTest : [task01][開始執行] 2019-11-30 14:03:35.828 INFO 64639 --- [ main] c.i.s.l.asynctask.service.DemoService : [execute01] 2019-11-30 14:03:45.833 INFO 64639 --- [ main] c.i.s.l.asynctask.service.DemoService : [execute02] 2019-11-30 14:03:50.834 INFO 64639 --- [ main] c.i.s.l.a.service.DemoServiceTest : [task01][結束執行，消耗時長 15014 毫秒]

• DemoService 的兩個方法，順序執行，一共消耗 15 秒左右。

• DemoService 的兩個方法，都在主線程中執行。

2.5 異步調用測試

修改 DemoService 的代碼，增加?#execute01()?和?#execute02()?的異步調用。代碼如下：

// DemoService.java@Async public Integer execute01Async() {return this.execute01(); }@Async public Integer execute02Async() {return this.execute02(); }

• 額外增加了?#execute01Async()?和?#execute02Async()?方法，主要是不想破壞上面的「2.4 同步調用測試」哈。實際上，可以在?#execute01()?和?#execute02()?方法上，添加?@Async?注解，實現異步調用。

修改?DemoServiceTest?測試類，編寫?#task02()?方法，異步調用上述的兩個方法。代碼如下：

// DemoServiceTest.java@Test public void task02() {long now = System.currentTimeMillis();logger.info("[task02][開始執行]");demoService.execute01Async();demoService.execute02Async();logger.info("[task02][結束執行，消耗時長 {} 毫秒]", System.currentTimeMillis() - now); }

運行單元測試，執行日志如下：

2019-11-30 15:57:45.809 INFO 69165 --- [ main] c.i.s.l.a.service.DemoServiceTest : [task02][開始執行] 2019-11-30 15:57:45.836 INFO 69165 --- [ main] c.i.s.l.a.service.DemoServiceTest : [task02][結束執行，消耗時長 27 毫秒]2019-11-30 15:57:45.844 INFO 69165 --- [ task-1] c.i.s.l.asynctask.service.DemoService : [execute01] 2019-11-30 15:57:45.844 INFO 69165 --- [ task-2] c.i.s.l.asynctask.service.DemoService : [execute02]

• DemoService 的兩個方法，異步執行，所以主線程只消耗 27 毫秒左右。注意，實際這兩個方法，并沒有執行完成。

• DemoService 的兩個方法，都在異步的線程池中，進行執行。

2.6 等待異步調用完成測試

在?「2.5 異步調用測試」?中，兩個方法只是發布異步調用，并未執行完成。在一些業務場景中，我們希望達到異步調用的效果，同時主線程阻塞等待異步調用的結果。

修改 DemoService 的代碼，增加?#execute01()?和?#execute02()?的異步調用，并返回 Future 對象。代碼如下：

// DemoService.java@Async public Future<Integer> execute01AsyncWithFuture() {return AsyncResult.forValue(this.execute01()); }@Async public Future<Integer> execute02AsyncWithFuture() {return AsyncResult.forValue(this.execute02()); }

• 相比?「2.5 異步調用測試」?的兩個方法，我們額外增加調用?AsyncResult#forValue(V value)?方法，返回帶有執行結果的 Future 對象。

修改?DemoServiceTest?測試類，編寫?#task03()?方法，異步調用上述的兩個方法，并阻塞等待執行完成。代碼如下：

// DemoServiceTest.java@Test public void task03() throws ExecutionException, InterruptedException {long now = System.currentTimeMillis();logger.info("[task03][開始執行]");// <1> 執行任務Future<Integer> execute01Result = demoService.execute01AsyncWithFuture();Future<Integer> execute02Result = demoService.execute02AsyncWithFuture();// <2> 阻塞等待結果execute01Result.get();execute02Result.get();logger.info("[task03][結束執行，消耗時長 {} 毫秒]", System.currentTimeMillis() - now); }

• <1>?處，異步調用兩個方法，并返回對應的 Future 對象。這樣，這兩個異步調用的邏輯，可以并行執行。

• <2>?處，分別調用兩個 Future 對象的?#get()?方法，阻塞等待結果。

運行單元測試，執行日志如下：

2019-11-30 16:10:22.226 INFO 69641 --- [ main] c.i.s.l.a.service.DemoServiceTest : [task03][開始執行]2019-11-30 16:10:22.272 INFO 69641 --- [ task-1] c.i.s.l.asynctask.service.DemoService : [execute01] 2019-11-30 16:10:22.272 INFO 69641 --- [ task-2] c.i.s.l.asynctask.service.DemoService : [execute02]2019-11-30 16:10:32.276 INFO 69641 --- [ main] c.i.s.l.a.service.DemoServiceTest : [task03][結束執行，消耗時長 10050 毫秒]

• DemoService 的兩個方法，異步執行，因為主線程阻塞等待執行結果，所以消耗 10 秒左右。當同時有多個異步調用，并阻塞等待執行結果，消耗時長由最慢的異步調用的邏輯所決定。

• DemoService 的兩個方法，都在異步的線程池中，進行執行。

下面「2.7 應用配置文件」小節，是補充知識，建議看看。

2.7 應用配置文件

在?application.yml?中，添加 Spring Task 定時任務的配置，如下：

spring:task:# Spring 執行器配置，對應 TaskExecutionProperties 配置類。對于 Spring 異步任務，會使用該執行器。execution:thread-name-prefix: task- # 線程池的線程名的前綴。默認為 task- ，建議根據自己應用來設置pool: # 線程池相關core-size: 8 # 核心線程數，線程池創建時候初始化的線程數。默認為 8 。max-size: 20 # 最大線程數，線程池最大的線程數，只有在緩沖隊列滿了之后，才會申請超過核心線程數的線程。默認為 Integer.MAX_VALUEkeep-alive: 60s # 允許線程的空閑時間，當超過了核心線程之外的線程，在空閑時間到達之后會被銷毀。默認為 60 秒queue-capacity: 200 # 緩沖隊列大小，用來緩沖執行任務的隊列的大小。默認為 Integer.MAX_VALUE 。allow-core-thread-timeout: true # 是否允許核心線程超時，即開啟線程池的動態增長和縮小。默認為 true 。shutdown:await-termination: true # 應用關閉時，是否等待定時任務執行完成。默認為 false ，建議設置為 trueawait-termination-period: 60 # 等待任務完成的最大時長，單位為秒。默認為 0 ，根據自己應用來設置

• 在?spring.task.execution?配置項，Spring Task 調度任務的配置，對應?TaskExecutionProperties?配置類。

• Spring Boot?TaskExecutionAutoConfiguration?自動化配置類，實現 Spring Task 的自動配置，創建?ThreadPoolTaskExecutor?基于線程池的任務執行器。本質上，ThreadPoolTaskExecutor 是基于 ThreadPoolExecutor 的封裝，主要增加提交任務，返回?ListenableFuture?對象的功能。

注意，spring.task.execution.shutdown?配置項，是為了實現 Spring Task 異步任務的優雅關閉。我們想象一下，如果異步任務在執行的過程中，如果應用開始關閉，把異步任務需要使用到的 Spring Bean 進行銷毀，例如說數據庫連接池，那么此時異步任務還在執行中，一旦需要訪問數據庫，可能會導致報錯。

• 所以，通過配置?await-termination = true?，實現應用關閉時，等待異步任務執行完成。這樣，應用在關閉的時，Spring 會優先等待 ThreadPoolTaskScheduler 執行完任務之后，再開始 Spring Bean 的銷毀。

• 同時，又考慮到我們不可能無限等待異步任務全部執行結束，因此可以配置?await-termination-period = 60?，等待任務完成的最大時長，單位為秒。具體設置多少的等待時長，可以根據自己應用的需要。

3. 異步回調

示例代碼對應倉庫：lab-29-async-demo?。

😈 異步 + 回調，快活似神仙。所以本小節我們來看看，如何在異步調用完成后，實現自定義回調。

考慮到讓胖友更加理解 Spring Task 異步回調是如何實現的，我們會在?「3.1 AsyncResult」?和?「3.2 ListenableFutureTask」小節進行部分源碼解析，請保持淡定。如果不想看的胖友，可以直接看?「3.3 具體示例」?小節。

友情提示：該示例，基于?「2. 快速入門」?的?lab-29-async-demo?的基礎上，繼續改造。

3.1 AsyncResult

在?「2.6 等待異步調用完成測試」?中，我們看到了?AsyncResult?類，表示異步結果。返回結果分成兩種情況：

• 執行成功時，調用?AsyncResult#forValue(V value)?靜態方法，返回成功的 ListenableFuture 對象。代碼如下：

  // AsyncResult.java@Nullableprivate final V value;public static <V> ListenableFuture<V> forValue(V value) {return new AsyncResult<>(value, null);}

• 執行異常時，調用?AsyncResult#forExecutionException(Throwable ex)?靜態方法，返回異常的 ListenableFuture 對象。代碼如下：

  // AsyncResult.java@Nullableprivate final Throwable executionException;public static <V> ListenableFuture<V> forExecutionException(Throwable ex) {return new AsyncResult<>(null, ex);}

同時，AsyncResult 實現了?ListenableFuture?接口，提供異步執行結果的回調處理。這里，我們先來看看 ListenableFuture 接口。代碼如下：

// ListenableFuture.javapublic interface ListenableFuture<T> extends Future<T> {// 添加回調方法，統一處理成功和異常的情況。void addCallback(ListenableFutureCallback<? super T> callback);// 添加成功和失敗的回調方法，分別處理成功和異常的情況。void addCallback(SuccessCallback<? super T> successCallback, FailureCallback failureCallback);// 將 ListenableFuture 轉換成 JDK8 提供的 CompletableFuture 。// 這樣，后續我們可以使用 ListenableFuture 來設置回調// 不了解 CompletableFuture 的胖友，可以看看 https://colobu.com/2016/02/29/Java-CompletableFuture/ 文章。default CompletableFuture<T> completable() {CompletableFuture<T> completable = new DelegatingCompletableFuture<>(this);addCallback(completable::complete, completable::completeExceptionally);return completable;}}

• 看下每個接口方法上的注釋。

因為 ListenableFuture 繼承了?Future?接口，所以 AsyncResult 也需要實現 Future 接口。這里，我們再來看看 Future 接口。代碼如下：

// Future.java public interface Future<V> {// 獲取異步執行的結果，如果沒有結果可用，此方法會阻塞直到異步計算完成。V get() throws InterruptedException, ExecutionException;// 獲取異步執行結果，如果沒有結果可用，此方法會阻塞，但是會有時間限制，如果阻塞時間超過設定的 timeout 時間，該方法將拋出異常。V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;// 如果任務執行結束，無論是正常結束或是中途取消還是發生異常，都返回 true 。boolean isDone();// 如果任務完成前被取消，則返回 true 。boolean isCancelled();// 如果任務還沒開始，執行 cancel(...) 方法將返回 false；// 如果任務已經啟動，執行 cancel(true) 方法將以中斷執行此任務線程的方式來試圖停止任務，如果停止成功，返回 true ；// 當任務已經啟動，執行c ancel(false) 方法將不會對正在執行的任務線程產生影響(讓線程正常執行到完成)，此時返回 false ；// 當任務已經完成，執行 cancel(...) 方法將返回 false 。// mayInterruptRunning 參數表示是否中斷執行中的線程。boolean cancel(boolean mayInterruptIfRunning);}

• 如上注釋內容，參考自?《Java 多線程編程：Callable、Future 和 FutureTask 淺析》?文章。

AsyncResult 對 ListenableFuture 定義的?#addCallback(...)?接口方法，實現代碼如下：

// AsyncResult.java@Override public void addCallback(ListenableFutureCallback<? super V> callback) {addCallback(callback, callback); }@Override public void addCallback(SuccessCallback<? super V> successCallback, FailureCallback failureCallback) {try {if (this.executionException != null) { // <1>failureCallback.onFailure(exposedException(this.executionException));} else { // <2>successCallback.onSuccess(this.value);}} catch (Throwable ex) { // <3>// Ignore} }// 從 ExecutionException 中，獲得原始異常。 private static Throwable exposedException(Throwable original) {if (original instanceof ExecutionException) {Throwable cause = original.getCause();if (cause != null) {return cause;}}return original; }

• ListenableFutureCallback?接口，同時繼承?SuccessCallback?和?FailureCallback?接口。

• <1>?處，如果是異常的結果，調用 FailureCallback 的回調。

• <2>?處，如果是正常的結果，調用 SuccessCallback 的回調。

• <3>?處，如果回調的邏輯發生異常，直接忽略。😈 所有，如果如果有多個回調，如果有一個回調發生異常，不會影響后續的回調。

(⊙o⊙)… 不過有點懵逼的是，不是應該在異步調用執行成功后，才進行回調么？！怎么這里一添加回調方法，就直接執行了？！不要著急，答案在?「3.2 ListenableFutureTask」?中解答。

實際上，AsyncResult 是作為異步執行的結果。既然是結果，執行就已經完成。所以，在我們調用?#addCallback(...)?接口方法來添加回調時，必然直接使用回調處理執行的結果。

AsyncResult 對 ListenableFuture 定義的?#completable(...)?接口方法，實現代碼如下：

// AsyncResult.java@Override public CompletableFuture<V> completable() {if (this.executionException != null) {CompletableFuture<V> completable = new CompletableFuture<>();completable.completeExceptionally(exposedException(this.executionException));return completable;} else {return CompletableFuture.completedFuture(this.value);} }

• 直接將結果包裝成 CompletableFuture 對象。

AsyncResult 對 Future 定義的所有方法，實現代碼如下：

// AsyncResult.java@Override public boolean cancel(boolean mayInterruptIfRunning) {return false; // 因為是 AsyncResult 是執行結果，所以直接返回 false 表示取消失敗。 }@Override public boolean isCancelled() {return false; // 因為是 AsyncResult 是執行結果，所以直接返回 false 表示未取消。 }@Override public boolean isDone() {return true; // 因為是 AsyncResult 是執行結果，所以直接返回 true 表示已完成。 }@Override @Nullable public V get() throws ExecutionException {// 如果發生異常，則拋出該異常。if (this.executionException != null) {throw (this.executionException instanceof ExecutionException ?(ExecutionException) this.executionException :new ExecutionException(this.executionException));}// 如果執行成功，則返回該 value 結果return this.value; }@Override @Nullable public V get(long timeout, TimeUnit unit) throws ExecutionException {return get(); }

• 胖友自己看看代碼上的注釋。

😈 看到這里，相信很多胖友會是一臉懵逼，淡定淡定。看源碼這個事兒，總是柳暗花明又一村。

3.2 ListenableFutureTask

在我們調用使用?@Async?注解的方法時，如果方法返回的類型是 ListenableFuture 的情況下，實際方法返回的是?ListenableFutureTask?對象。

感興趣的胖友，可以看看?AsyncExecutionInterceptor?類、《Spring 異步調用原理及Spring AOP 攔截器鏈原理》?文章。

ListenableFutureTask 類，也實現 ListenableFuture 接口，繼承?FutureTask?類，ListenableFuture 的 FutureTask 實現類。

ListenableFutureTask 對 ListenableFuture 定義的?#addCallback(...)?方法，實現代碼如下：

// ListenableFutureTask.javaprivate final ListenableFutureCallbackRegistry<T> callbacks = new ListenableFutureCallbackRegistry<T>();@Override public void addCallback(ListenableFutureCallback<? super T> callback) {this.callbacks.addCallback(callback); }@Override public void addCallback(SuccessCallback<? super T> successCallback, FailureCallback failureCallback) {this.callbacks.addSuccessCallback(successCallback);this.callbacks.addFailureCallback(failureCallback); }

• 暫存回調到?ListenableFutureCallbackRegistry?中先。😈 這樣看起來，和我們想象中的異步回調有點像了。

ListenableFutureTask 對 FutureTask 已實現的?#done()?方法，進行重寫。實現代碼如下：

// ListenableFutureTask.java@Override protected void done() {Throwable cause;try {// <1> 獲得執行結果T result = get();// <2.1> 執行成功，執行成功的回調this.callbacks.success(result);return;} catch (InterruptedException ex) { // 如果有中斷異常 InterruptedException ，則打斷當前線程，并直接返回Thread.currentThread().interrupt();return;} catch (ExecutionException ex) { // 如果有 ExecutionException 異常，獲得其真實的異常，并設置到 cause 中cause = ex.getCause();if (cause == null) {cause = ex;}} catch (Throwable ex) { // 設置異常到 cause 中cause = ex;}// 執行異常，執行異常的回調this.callbacks.failure(cause); }

• <1>?處，調用?#get()?方法，獲得執行結果。

• <2.1>?處，執行成功，執行成功的回調。

• <2.2>?處，執行異常，執行異常的回調。

這樣一看，是不是對 AsyncResult 和 ListenableFutureTask 就有點感覺了。

3.3 具體示例

下面，讓我們來寫一個異步回調的示例。修改 DemoService 的代碼，增加?#execute02()?的異步調用，并返回 ListenableFuture 對象。代碼如下：

// DemoService.java@Async public ListenableFuture<Integer> execute01AsyncWithListenableFuture() {try {return AsyncResult.forValue(this.execute02());} catch (Throwable ex) {return AsyncResult.forExecutionException(ex);} }

• 根據執行的結果，包裝出成功還是異常的 AsyncResult 對象。

修改?DemoServiceTest?測試類，編寫?#task04()?方法，異步調用上述的方法，在塞等待執行完成的同時，添加相應的回調 Callback 方法。代碼如下：

// DemoServiceTest.java@Test public void task04() throws ExecutionException, InterruptedException {long now = System.currentTimeMillis();logger.info("[task04][開始執行]");// <1> 執行任務ListenableFuture<Integer> execute01Result = demoService.execute01AsyncWithListenableFuture();logger.info("[task04][execute01Result 的類型是：({})]",execute01Result.getClass().getSimpleName());execute01Result.addCallback(new SuccessCallback<Integer>() { // <2.1> 增加成功的回調@Overridepublic void onSuccess(Integer result) {logger.info("[onSuccess][result: {}]", result);}}, new FailureCallback() { // <2.1> 增加失敗的回調@Overridepublic void onFailure(Throwable ex) {logger.info("[onFailure][發生異常]", ex);}});execute01Result.addCallback(new ListenableFutureCallback<Integer>() { // <2.2> 增加成功和失敗的統一回調@Overridepublic void onSuccess(Integer result) {logger.info("[onSuccess][result: {}]", result);}@Overridepublic void onFailure(Throwable ex) {logger.info("[onFailure][發生異常]", ex);}});// <3> 阻塞等待結果execute01Result.get();logger.info("[task04][結束執行，消耗時長 {} 毫秒]", System.currentTimeMillis() - now); }

• <1>?處，調用?DemoService#execute01AsyncWithListenableFuture()?方法，異步調用該方法，并返回 ListenableFutureTask 對象。這里，我們看下打印的日志。

  2019-11-30 19:17:51.320 INFO 77624 --- [ main] c.i.s.l.a.service.DemoServiceTest : [task04][execute01Result 的類型是：(ListenableFutureTask)]

• <2.1>?處，增加成功的回調和失敗的回調。

• <2.2>?處，增加成功和失敗的統一回調。

• <3>?處，阻塞等待結果。執行完成后，我們會看到回調被執行，打印日志如下：

  2019-11-30 19:17:56.330 INFO 77624 --- [ task-1] c.i.s.l.a.service.DemoServiceTest : [onSuccess][result: 2]2019-11-30 19:17:56.331 INFO 77624 --- [ task-1] c.i.s.l.a.service.DemoServiceTest : [onSuccess][result: 2]

4. 異步異常處理器

示例代碼對應倉庫：lab-29-async-demo?。

在?《芋道 Spring Boot SpringMVC 入門》?的?「5. 全局異常處理」?中，我們實現了對 SpringMVC 請求異常的全局處理。那么，Spring Task 異步調用異常是否有全局處理呢？答案是有，通過實現?AsyncUncaughtExceptionHandler?接口，達到對異步調用的異常的統一處理。

友情提示：該示例，基于?「2. 快速入門」?的?lab-29-async-demo?的基礎上，繼續改造。

4.1 GlobalAsyncExceptionHandler

在?cn.iocoder.springboot.lab29.asynctask.core.async?包路徑，創建?GlobalAsyncExceptionHandler?類，全局統一的異步調用異常的處理器。代碼如下：

// GlobalAsyncExceptionHandler.java@Component public class GlobalAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {private Logger logger = LoggerFactory.getLogger(getClass());@Overridepublic void handleUncaughtException(Throwable ex, Method method, Object... params) {logger.error("[handleUncaughtException][method({}) params({}) 發生異常]",method, params, ex);}}

• 類上，我們添加了?@Component?注解，考慮到胖友可能會注入一些 Spring Bean 到屬性中。

• 實現?#handleUncaughtException(Throwable ex, Method method, Object... params)?方法，打印異常日志。😈 這樣，后續如果我們接入 ELK ，就可以基于該異常日志進行告警。

注意，AsyncUncaughtExceptionHandler 只能攔截返回類型非 Future?的異步調用方法。通過看?AsyncExecutionAspectSupport#handleError(Throwable ex, Method method, Object... params)?的源碼，可以很容易得到這個結論，代碼如下：

// AsyncExecutionAspectSupport.javaprotected void handleError(Throwable ex, Method method, Object... params) throws Exception {// 重點！！！如果返回類型是 Future ，則直接拋出該異常。if (Future.class.isAssignableFrom(method.getReturnType())) {ReflectionUtils.rethrowException(ex);} else {// 否則，交給 AsyncUncaughtExceptionHandler 來處理。// Could not transmit the exception to the caller with default executortry {this.exceptionHandler.obtain().handleUncaughtException(ex, method, params);} catch (Throwable ex2) {logger.warn("Exception handler for async method '" + method.toGenericString() +"' threw unexpected exception itself", ex2);}} }

• 對了，AsyncExecutionAspectSupport 是 AsyncExecutionInterceptor 的父類喲。

所以喲，返回類型為 Future 的異步調用方法，需要通過「3. 異步回調」來處理。

4.2 AsyncConfig

在?cn.iocoder.springboot.lab29.asynctask.config?包路徑，創建?AsyncConfig?類，配置異常處理器。代碼如下：

// AsyncConfig.java@Configuration @EnableAsync // 開啟 @Async 的支持 public class AsyncConfig implements AsyncConfigurer {@Autowiredprivate GlobalAsyncExceptionHandler exceptionHandler;@Overridepublic Executor getAsyncExecutor() {return null;}@Overridepublic AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {return exceptionHandler;}}

• 在類上添加?@EnableAsync?注解，啟用異步功能。這樣「2. Application」?的?@EnableAsync?注解，也就可以去掉了。

• 實現?AsyncConfigurer?接口，實現異步相關的全局配置。😈 此時此刻，胖友有沒想到 SpringMVC 的?WebMvcConfigurer?接口。

• 實現?#getAsyncUncaughtExceptionHandler()?方法，返回我們定義的 GlobalAsyncExceptionHandler 對象。

• 實現?#getAsyncExecutor()?方法，返回 Spring Task 異步任務的默認執行器。這里，我們返回了?null?，并未定義默認執行器。所以最終會使用?TaskExecutionAutoConfiguration?自動化配置類創建出來的 ThreadPoolTaskExecutor 任務執行器，作為默認執行器。

4.3 DemoService

修改 DemoService 的代碼，增加?#zhaoDaoNvPengYou(...)?的異步調用。代碼如下：

// DemoService.java@Async public Integer zhaoDaoNvPengYou(Integer a, Integer b) {throw new RuntimeException("程序員不需要女朋友"); }

• 直接給想要找女朋友的程序員，拋出該異常。

4.4 簡單測試

修改?DemoServiceTest?測試類，編寫?#testZhaoDaoNvPengYou()?方法，異步調用上述的方法。代碼如下：

// DemoServiceTest.java@Test public void testZhaoDaoNvPengYou() throws InterruptedException {demoService.zhaoDaoNvPengYou(1, 2);// sleep 1 秒，保證異步調用的執行Thread.sleep(1000); }

運行單元測試，執行日志如下：

2019-11-30 09:22:52.962 ERROR 86590 --- [ task-1] .i.s.l.a.c.a.GlobalAsyncExceptionHandler : [handleUncaughtException][method(public java.lang.Integer cn.iocoder.springboot.lab29.asynctask.service.DemoService.zhaoDaoNvPengYou(java.lang.Integer,java.lang.Integer)) params([1, 2]) 發生異常]java.lang.RuntimeException: 程序員不需要女朋友

• 😈 異步調用的異常成功被 GlobalAsyncExceptionHandler 攔截。

5. 自定義執行器

示例代碼對應倉庫：lab-29-async-two?。

在?「2. 快速入門」?中，我們使用 Spring Boot?TaskExecutionAutoConfiguration?自動化配置類，實現自動配置 ThreadPoolTaskExecutor 任務執行器。

本小節，我們希望兩個自定義 ThreadPoolTaskExecutor 任務執行器，實現不同方法，分別使用這兩個 ThreadPoolTaskExecutor 任務執行器。

友情提示：考慮到不破壞上面入門的示例，所以我們新建了?lab-29-async-two?項目。

5.1 引入依賴

在?pom.xml?文件中，引入相關依賴。

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.2.1.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><modelVersion>4.0.0</modelVersion><artifactId>lab-29-async-demo</artifactId><dependencies><!-- 引入 Spring Boot 依賴 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><!-- 方便等會寫單元測試 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies></project>

• 和?「2.1 引入依賴」?一致。

5.2 應用配置文件

在?application.yml?中，添加 Spring Task 定時任務的配置，如下：

spring:task:# Spring 執行器配置，對應 TaskExecutionProperties 配置類。對于 Spring 異步任務，會使用該執行器。execution-one:thread-name-prefix: task-one- # 線程池的線程名的前綴。默認為 task- ，建議根據自己應用來設置pool: # 線程池相關core-size: 8 # 核心線程數，線程池創建時候初始化的線程數。默認為 8 。max-size: 20 # 最大線程數，線程池最大的線程數，只有在緩沖隊列滿了之后，才會申請超過核心線程數的線程。默認為 Integer.MAX_VALUEkeep-alive: 60s # 允許線程的空閑時間，當超過了核心線程之外的線程，在空閑時間到達之后會被銷毀。默認為 60 秒queue-capacity: 200 # 緩沖隊列大小，用來緩沖執行任務的隊列的大小。默認為 Integer.MAX_VALUE 。allow-core-thread-timeout: true # 是否允許核心線程超時，即開啟線程池的動態增長和縮小。默認為 true 。shutdown:await-termination: true # 應用關閉時，是否等待定時任務執行完成。默認為 false ，建議設置為 trueawait-termination-period: 60 # 等待任務完成的最大時長，單位為秒。默認為 0 ，根據自己應用來設置# Spring 執行器配置，對應 TaskExecutionProperties 配置類。對于 Spring 異步任務，會使用該執行器。execution-two:thread-name-prefix: task-two- # 線程池的線程名的前綴。默認為 task- ，建議根據自己應用來設置pool: # 線程池相關core-size: 8 # 核心線程數，線程池創建時候初始化的線程數。默認為 8 。max-size: 20 # 最大線程數，線程池最大的線程數，只有在緩沖隊列滿了之后，才會申請超過核心線程數的線程。默認為 Integer.MAX_VALUEkeep-alive: 60s # 允許線程的空閑時間，當超過了核心線程之外的線程，在空閑時間到達之后會被銷毀。默認為 60 秒queue-capacity: 200 # 緩沖隊列大小，用來緩沖執行任務的隊列的大小。默認為 Integer.MAX_VALUE 。allow-core-thread-timeout: true # 是否允許核心線程超時，即開啟線程池的動態增長和縮小。默認為 true 。shutdown:await-termination: true # 應用關閉時，是否等待定時任務執行完成。默認為 false ，建議設置為 trueawait-termination-period: 60 # 等待任務完成的最大時長，單位為秒。默認為 0 ，根據自己應用來設置

• 在?spring.task?配置項下，我們新增了?execution-one?和?execution-two?兩個執行器的配置。在格式上，我們保持和在「2.7 應用配置文件」看到的?spring.task.exeuction?一致，方便我們后續復用 TaskExecutionProperties 屬性配置類來映射。

5.3 AsyncConfig

在?cn.iocoder.springboot.lab29.asynctask.config?包路徑，創建?AsyncConfig?類，配置兩個執行器。代碼如下：

// AsyncConfig.java@Configuration @EnableAsync // 開啟 @Async 的支持 public class AsyncConfig {public static final String EXECUTOR_ONE_BEAN_NAME = "executor-one";public static final String EXECUTOR_TWO_BEAN_NAME = "executor-two";@Configurationpublic static class ExecutorOneConfiguration {@Bean(name = EXECUTOR_ONE_BEAN_NAME + "-properties")@Primary@ConfigurationProperties(prefix = "spring.task.execution-one") // 讀取 spring.task.execution-one 配置到 TaskExecutionProperties 對象public TaskExecutionProperties taskExecutionProperties() {return new TaskExecutionProperties();}@Bean(name = EXECUTOR_ONE_BEAN_NAME)public ThreadPoolTaskExecutor threadPoolTaskExecutor() {// 創建 TaskExecutorBuilder 對象TaskExecutorBuilder builder = createTskExecutorBuilder(this.taskExecutionProperties());// 創建 ThreadPoolTaskExecutor 對象return builder.build();}}@Configurationpublic static class ExecutorTwoConfiguration {@Bean(name = EXECUTOR_TWO_BEAN_NAME + "-properties")@ConfigurationProperties(prefix = "spring.task.execution-two") // 讀取 spring.task.execution-two 配置到 TaskExecutionProperties 對象public TaskExecutionProperties taskExecutionProperties() {return new TaskExecutionProperties();}@Bean(name = EXECUTOR_TWO_BEAN_NAME)public ThreadPoolTaskExecutor threadPoolTaskExecutor() {// 創建 TaskExecutorBuilder 對象TaskExecutorBuilder builder = createTskExecutorBuilder(this.taskExecutionProperties());// 創建 ThreadPoolTaskExecutor 對象return builder.build();}}private static TaskExecutorBuilder createTskExecutorBuilder(TaskExecutionProperties properties) {// Pool 屬性TaskExecutionProperties.Pool pool = properties.getPool();TaskExecutorBuilder builder = new TaskExecutorBuilder();builder = builder.queueCapacity(pool.getQueueCapacity());builder = builder.corePoolSize(pool.getCoreSize());builder = builder.maxPoolSize(pool.getMaxSize());builder = builder.allowCoreThreadTimeOut(pool.isAllowCoreThreadTimeout());builder = builder.keepAlive(pool.getKeepAlive());// Shutdown 屬性TaskExecutionProperties.Shutdown shutdown = properties.getShutdown();builder = builder.awaitTermination(shutdown.isAwaitTermination());builder = builder.awaitTerminationPeriod(shutdown.getAwaitTerminationPeriod());// 其它基本屬性builder = builder.threadNamePrefix(properties.getThreadNamePrefix()); // builder = builder.customizers(taskExecutorCustomizers.orderedStream()::iterator); // builder = builder.taskDecorator(taskDecorator.getIfUnique());return builder;}}

• 參考 Spring Boot?TaskExecutionAutoConfiguration?自動化配置類，我們創建了 ExecutorOneConfiguration 和 ExecutorTwoConfiguration 配置類，來分別創建 Bean 名字為?executor-one?和?executor-two?兩個執行器。

5.4 DemoService

在?cn.iocoder.springboot.lab29.asynctask.service?包路徑下，創建?DemoService?類。代碼如下：

// DemoService.java@Service public class DemoService {private Logger logger = LoggerFactory.getLogger(getClass());@Async(AsyncConfig.EXECUTOR_ONE_BEAN_NAME)public Integer execute01() {logger.info("[execute01]");return 1;}@Async(AsyncConfig.EXECUTOR_TWO_BEAN_NAME)public Integer execute02() {logger.info("[execute02]");return 2;}}

• 在?@Async?注解上，我們設置了其使用的執行器的 Bean 名字。

5.5 簡單測試

創建?DemoServiceTest?測試類，編寫?#testExecute()?方法，異步調用 DemoService 的上述兩個方法。代碼如下：

// DemoServiceTest.java@RunWith(SpringRunner.class) @SpringBootTest(classes = Application.class) public class DemoServiceTest {@Autowiredprivate DemoService demoService;@Testpublic void testExecute() throws InterruptedException {demoService.execute01();demoService.execute02();// sleep 1 秒，保證異步調用的執行Thread.sleep(1000);}}

運行單元測試，執行日志如下：

2019-11-30 10:25:53.068 INFO 89290 --- [ task-one-1] c.i.s.l.asynctask.service.DemoService : [execute01] 2019-11-30 10:25:53.068 INFO 89290 --- [ task-two-1] c.i.s.l.asynctask.service.DemoService : [execute02]

• 從日志中，我們可以看到，#execute01()?方法在?executor-one?執行器中執行，而?#execute02()?方法在?executor-two?執行器中執行。符合預期~

666. 彩蛋

😈 發現自己真是一個啰嗦的老男孩，挺簡單一東西，結果又寫了老長一篇。不過最后還是要嘮叨下，如果胖友使用 Spring Task 的異步任務，一定要注意兩個點：

• JVM 應用的正常優雅關閉，保證異步任務都被執行完成。

• 編寫異步異常處理器 GlobalAsyncExceptionHandler ，記錄異常日志，進行監控告警。

嗯~~~如果覺得不過癮的胖友，可以再去看看?《Spring Framework Documentation —— Task Execution and Scheduling》?文檔。

不過呢，Spring Task 異步任務，在項目中使用的并不多，更多的選擇，還是可靠的分布式隊列，嘿嘿。當然，艿艿在自己的開源項目?onemall?中，使用?AccessLogInterceptor?攔截器，記錄訪問日志到數據庫。因為訪問日志更多是用于監控和排查問題，所以即使有一定的丟失，影響也不大。

總結

以上是生活随笔為你收集整理的Spring 异步调用，一行代码实现！舒服，不接受任何反驳~的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。