javascript
Spring 异步调用,一行代码实现!舒服,不接受任何反驳~
本文在提供完整代碼示例,可見?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 异步调用,一行代码实现!舒服,不接受任何反驳~的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 琢磨琢磨,while (true) 和
- 下一篇: Java 最坑爹的 10 大功能点!