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

原創(chuàng)不易，給點個?Star?嘿，一起沖鴨！

1. 概述

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

異步調(diào)用，對應的是同步調(diào)用。

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

• 異步調(diào)用：指程序在順序執(zhí)行時，不等待異步調(diào)用的語句返回結(jié)果，就執(zhí)行后面的程序。

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

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

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

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

所以，使用進程內(nèi)的隊列或者線程池來實現(xiàn)異步調(diào)用的話，一定要盡可能的保證 JVM 進程的優(yōu)雅關(guān)閉，保證它們在關(guān)閉前被執(zhí)行完成。

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

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

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

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

2. 快速入門

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

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

2.1 引入依賴

在?pom.xml?文件中，引入相關(guān)依賴。

<?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

創(chuàng)建?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?包路徑下，創(chuàng)建?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?打印日志，方便我們看到每個方法的開始執(zhí)行時間，和執(zhí)行所在線程。

2.4 同步調(diào)用測試

創(chuàng)建?DemoServiceTest?測試類，編寫?#task01()?方法，同步調(diào)用 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][開始執(zhí)行]");demoService.execute01();demoService.execute02();logger.info("[task01][結(jié)束執(zhí)行，消耗時長 {} 毫秒]", System.currentTimeMillis() - now);}}

運行單元測試，執(zhí)行日志如下：

2019-11-30 14:03:35.820 INFO 64639 --- [ main] c.i.s.l.a.service.DemoServiceTest : [task01][開始執(zhí)行] 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][結(jié)束執(zhí)行，消耗時長 15014 毫秒]

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

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

2.5 異步調(diào)用測試

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

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

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

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

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

運行單元測試，執(zhí)行日志如下：

2019-11-30 15:57:45.809 INFO 69165 --- [ main] c.i.s.l.a.service.DemoServiceTest : [task02][開始執(zhí)行] 2019-11-30 15:57:45.836 INFO 69165 --- [ main] c.i.s.l.a.service.DemoServiceTest : [task02][結(jié)束執(zhí)行，消耗時長 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 的兩個方法，異步執(zhí)行，所以主線程只消耗 27 毫秒左右。注意，實際這兩個方法，并沒有執(zhí)行完成。

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

2.6 等待異步調(diào)用完成測試

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

修改 DemoService 的代碼，增加?#execute01()?和?#execute02()?的異步調(diào)用，并返回 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 異步調(diào)用測試」?的兩個方法，我們額外增加調(diào)用?AsyncResult#forValue(V value)?方法，返回帶有執(zhí)行結(jié)果的 Future 對象。

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

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

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

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

運行單元測試，執(zhí)行日志如下：

2019-11-30 16:10:22.226 INFO 69641 --- [ main] c.i.s.l.a.service.DemoServiceTest : [task03][開始執(zhí)行]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][結(jié)束執(zhí)行，消耗時長 10050 毫秒]

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

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

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

2.7 應用配置文件

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

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

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

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

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

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

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

3. 異步回調(diào)

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

😈 異步 + 回調(diào)，快活似神仙。所以本小節(jié)我們來看看，如何在異步調(diào)用完成后，實現(xiàn)自定義回調(diào)。

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

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

3.1 AsyncResult

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

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

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

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

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

同時，AsyncResult 實現(xiàn)了?ListenableFuture?接口，提供異步執(zhí)行結(jié)果的回調(diào)處理。這里，我們先來看看 ListenableFuture 接口。代碼如下：

// ListenableFuture.javapublic interface ListenableFuture<T> extends Future<T> {// 添加回調(diào)方法，統(tǒng)一處理成功和異常的情況。void addCallback(ListenableFutureCallback<? super T> callback);// 添加成功和失敗的回調(diào)方法，分別處理成功和異常的情況。void addCallback(SuccessCallback<? super T> successCallback, FailureCallback failureCallback);// 將 ListenableFuture 轉(zhuǎn)換成 JDK8 提供的 CompletableFuture 。// 這樣，后續(xù)我們可以使用 ListenableFuture 來設(shè)置回調(diào)// 不了解 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 也需要實現(xiàn) Future 接口。這里，我們再來看看 Future 接口。代碼如下：

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

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

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

// 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>?處，如果是異常的結(jié)果，調(diào)用 FailureCallback 的回調(diào)。

• <2>?處，如果是正常的結(jié)果，調(diào)用 SuccessCallback 的回調(diào)。

• <3>?處，如果回調(diào)的邏輯發(fā)生異常，直接忽略。😈 所有，如果如果有多個回調(diào)，如果有一個回調(diào)發(fā)生異常，不會影響后續(xù)的回調(diào)。

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

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

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

// 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);} }

• 直接將結(jié)果包裝成 CompletableFuture 對象。

AsyncResult 對 Future 定義的所有方法，實現(xiàn)代碼如下：

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

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

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

3.2 ListenableFutureTask

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

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

ListenableFutureTask 類，也實現(xiàn) ListenableFuture 接口，繼承?FutureTask?類，ListenableFuture 的 FutureTask 實現(xiàn)類。

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

// 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); }

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

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

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

• <1>?處，調(diào)用?#get()?方法，獲得執(zhí)行結(jié)果。

• <2.1>?處，執(zhí)行成功，執(zhí)行成功的回調(diào)。

• <2.2>?處，執(zhí)行異常，執(zhí)行異常的回調(diào)。

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

3.3 具體示例

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

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

• 根據(jù)執(zhí)行的結(jié)果，包裝出成功還是異常的 AsyncResult 對象。

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

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

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

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

• <2.1>?處，增加成功的回調(diào)和失敗的回調(diào)。

• <2.2>?處，增加成功和失敗的統(tǒng)一回調(diào)。

• <3>?處，阻塞等待結(jié)果。執(zhí)行完成后，我們會看到回調(diào)被執(zhí)行，打印日志如下：

  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. 全局異常處理」?中，我們實現(xiàn)了對 SpringMVC 請求異常的全局處理。那么，Spring Task 異步調(diào)用異常是否有全局處理呢？答案是有，通過實現(xiàn)?AsyncUncaughtExceptionHandler?接口，達到對異步調(diào)用的異常的統(tǒng)一處理。

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

4.1 GlobalAsyncExceptionHandler

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

// 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({}) 發(fā)生異常]",method, params, ex);}}

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

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

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

// 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 的異步調(diào)用方法，需要通過「3. 異步回調(diào)」來處理。

4.2 AsyncConfig

在?cn.iocoder.springboot.lab29.asynctask.config?包路徑，創(chuàng)建?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?注解，也就可以去掉了。

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

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

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

4.3 DemoService

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

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

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

4.4 簡單測試

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

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

運行單元測試，執(zhí)行日志如下：

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]) 發(fā)生異常]java.lang.RuntimeException: 程序員不需要女朋友

• 😈 異步調(diào)用的異常成功被 GlobalAsyncExceptionHandler 攔截。

5. 自定義執(zhí)行器

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

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

本小節(jié)，我們希望兩個自定義 ThreadPoolTaskExecutor 任務執(zhí)行器，實現(xiàn)不同方法，分別使用這兩個 ThreadPoolTaskExecutor 任務執(zhí)行器。

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

5.1 引入依賴

在?pom.xml?文件中，引入相關(guān)依賴。

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

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

5.3 AsyncConfig

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

// 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() {// 創(chuàng)建 TaskExecutorBuilder 對象TaskExecutorBuilder builder = createTskExecutorBuilder(this.taskExecutionProperties());// 創(chuàng)建 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() {// 創(chuàng)建 TaskExecutorBuilder 對象TaskExecutorBuilder builder = createTskExecutorBuilder(this.taskExecutionProperties());// 創(chuàng)建 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?自動化配置類，我們創(chuàng)建了 ExecutorOneConfiguration 和 ExecutorTwoConfiguration 配置類，來分別創(chuàng)建 Bean 名字為?executor-one?和?executor-two?兩個執(zhí)行器。

5.4 DemoService

在?cn.iocoder.springboot.lab29.asynctask.service?包路徑下，創(chuàng)建?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?注解上，我們設(shè)置了其使用的執(zhí)行器的 Bean 名字。

5.5 簡單測試

創(chuàng)建?DemoServiceTest?測試類，編寫?#testExecute()?方法，異步調(diào)用 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 秒，保證異步調(diào)用的執(zhí)行Thread.sleep(1000);}}

運行單元測試，執(zhí)行日志如下：

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?執(zhí)行器中執(zhí)行，而?#execute02()?方法在?executor-two?執(zhí)行器中執(zhí)行。符合預期~

666. 彩蛋

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

• JVM 應用的正常優(yōu)雅關(guān)閉，保證異步任務都被執(zhí)行完成。

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

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

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。