h5網(wǎng)站建設(shè)方案,有視頻做的很好的網(wǎng)站嗎,萬網(wǎng)建站,安徽城鄉(xiāng)建設(shè)廳網(wǎng)站證件在Java開發(fā)領(lǐng)域#xff0c;異步編程是提升系統(tǒng)吞吐量、優(yōu)化用戶體驗的核心手段之一。而Spring框架提供的Async注解#xff0c;更是讓開發(fā)者無需深入了解復(fù)雜的線程池原理#xff0c;就能輕松實現(xiàn)異步調(diào)用。但實際開發(fā)中#xff0c;很多同學(xué)在使用Async時會遇到“異步不生效…在Java開發(fā)領(lǐng)域異步編程是提升系統(tǒng)吞吐量、優(yōu)化用戶體驗的核心手段之一。而Spring框架提供的Async注解更是讓開發(fā)者無需深入了解復(fù)雜的線程池原理就能輕松實現(xiàn)異步調(diào)用。但實際開發(fā)中很多同學(xué)在使用Async時會遇到“異步不生效”“線程池耗盡”“事務(wù)失效”等問題。本文將從實戰(zhàn)出發(fā)結(jié)合底層源碼全面拆解Async注解的使用方法、核心原理、避坑要點搭配可直接運行的實例讓你真正吃透異步編程的精髓。一、Async注解核心認(rèn)知什么是異步調(diào)用為什么需要它1.1 同步VS異步本質(zhì)區(qū)別在講解Async之前我們先明確同步調(diào)用與異步調(diào)用的核心差異同步調(diào)用方法A調(diào)用方法B后必須等待方法B執(zhí)行完畢并返回結(jié)果A才能繼續(xù)執(zhí)行整個過程是阻塞的。異步調(diào)用方法A調(diào)用方法B后無需等待B執(zhí)行完畢A可以直接繼續(xù)執(zhí)行后續(xù)邏輯而B會在獨立的線程中異步執(zhí)行。1.2 異步調(diào)用的適用場景異步調(diào)用適合處理“耗時且非核心流程”的操作典型場景包括接口響應(yīng)后的日志記錄、數(shù)據(jù)統(tǒng)計如用戶登錄后記錄登錄日志郵件/短信發(fā)送無需等待發(fā)送結(jié)果返回給前端大文件導(dǎo)出、數(shù)據(jù)批量處理避免阻塞主線程導(dǎo)致接口超時第三方接口調(diào)用如調(diào)用支付回調(diào)接口無需同步等待結(jié)果。1.3 Async的核心作用Spring的Async注解基于AOP實現(xiàn)通過動態(tài)代理機制將被注解的方法封裝到獨立的線程中執(zhí)行從而實現(xiàn)異步調(diào)用。其核心價值在于簡化異步編程無需手動創(chuàng)建線程池、管理線程生命周期解耦線程管理與業(yè)務(wù)邏輯開發(fā)者只需關(guān)注業(yè)務(wù)實現(xiàn)線程池配置統(tǒng)一管理支持靈活配置可自定義線程池參數(shù)、異常處理機制。二、Async基礎(chǔ)使用從環(huán)境搭建到第一個異步程序2.1 環(huán)境依賴準(zhǔn)備Maven使用Async需依賴Spring核心包結(jié)合實戰(zhàn)場景我們搭建一個Spring Boot項目核心依賴如下所有版本采用最新穩(wěn)定版dependencies !-- Spring Boot核心依賴 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId version3.2.5/version /dependency !-- Spring Boot測試依賴 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId version3.2.5/version scopetest/scope /dependency !-- Lombok簡化日志、Getter/Setter等 -- dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId version1.18.30/version scopeprovided/scope /dependency !-- FastJSON2JSON處理 -- dependency groupIdcom.alibaba.fastjson2/groupId artifactIdfastjson2/artifactId version2.0.49/version /dependency !-- Guava集合工具類 -- dependency groupIdcom.google.guava/groupId artifactIdguava/artifactId version33.2.1-jre/version /dependency !-- MyBatis-Plus持久層框架 -- dependency groupIdcom.baomidou/groupId artifactIdmybatis-plus-boot-starter/artifactId version3.5.5/version /dependency !-- MySQL驅(qū)動 -- dependency groupIdcom.mysql/groupId artifactIdmysql-connector-j/artifactId version8.3.0/version scoperuntime/scope /dependency !-- Swagger3接口文檔 -- dependency groupIdorg.springdoc/groupId artifactIdspringdoc-openapi-starter-webmvc-ui/artifactId version2.5.0/version /dependency /dependencies2.2 啟用AsyncEnableAsync注解要讓Spring識別Async注解必須在配置類或啟動類上添加EnableAsync注解該注解的作用是開啟Spring的異步方法支持底層會注冊異步方法處理器。package com.jam.demo; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.scheduling.annotation.EnableAsync; /** * 應(yīng)用啟動類 * 開啟異步支持EnableAsync * author ken */ SpringBootApplication EnableAsync public class AsyncDemoApplication { public static void main(String[] args) { SpringApplication.run(AsyncDemoApplication.class, args); } }2.3 第一個異步程序基礎(chǔ)使用示例2.3.1 異步服務(wù)接口與實現(xiàn)定義異步服務(wù)接口在實現(xiàn)類的方法上添加Async注解標(biāo)記該方法為異步方法。package com.jam.demo.service; /** * 異步服務(wù)接口 * author ken */ public interface AsyncService { /** * 基礎(chǔ)異步方法無返回值 * param taskName 任務(wù)名稱 */ void basicAsyncTask(String taskName); /** * 異步方法有返回值返回Future * param taskName 任務(wù)名稱 * param sleepTime 模擬耗時時間毫秒 * return 任務(wù)執(zhí)行結(jié)果 */ FutureString asyncTaskWithReturn(String taskName, long sleepTime); }package com.jam.demo.service.impl; import com.jam.demo.service.AsyncService; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import java.util.concurrent.Future; import java.util.concurrent.FutureTask; /** * 異步服務(wù)實現(xiàn)類 * author ken */ Service Slf4j public class AsyncServiceImpl implements AsyncService { /** * 基礎(chǔ)異步方法無返回值 * Async標(biāo)記該方法為異步方法使用默認(rèn)線程池 * param taskName 任務(wù)名稱 */ Override Async public void basicAsyncTask(String taskName) { log.info(【異步任務(wù)】{} 開始執(zhí)行當(dāng)前線程{}, taskName, Thread.currentThread().getName()); // 模擬耗時操作如日志記錄、郵件發(fā)送 try { Thread.sleep(2000); } catch (InterruptedException e) { log.error(【異步任務(wù)】{} 執(zhí)行異常, taskName, e); Thread.currentThread().interrupt(); } log.info(【異步任務(wù)】{} 執(zhí)行完畢, taskName); } /** * 異步方法有返回值 * 注意有返回值的異步方法必須返回Future或其實現(xiàn)類如FutureTask * param taskName 任務(wù)名稱 * param sleepTime 模擬耗時時間毫秒 * return 任務(wù)執(zhí)行結(jié)果 */ Override Async public FutureString asyncTaskWithReturn(String taskName, long sleepTime) { log.info(【異步任務(wù)有返回值】{} 開始執(zhí)行當(dāng)前線程{}預(yù)計耗時{}ms, taskName, Thread.currentThread().getName(), sleepTime); try { Thread.sleep(sleepTime); String result taskName 執(zhí)行成功; return new FutureTask(() - result); } catch (InterruptedException e) { log.error(【異步任務(wù)有返回值】{} 執(zhí)行異常, taskName, e); Thread.currentThread().interrupt(); return new FutureTask(() - taskName 執(zhí)行失敗); } } }2.3.2 測試接口驗證異步效果編寫Controller層接口調(diào)用異步服務(wù)方法驗證異步執(zhí)行效果。package com.jam.demo.controller; import com.jam.demo.service.AsyncService; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.tags.Tag; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import java.util.concurrent.Future; /** * 異步測試控制器 * author ken */ RestController RequestMapping(/async) Slf4j Tag(name 異步測試接口, description 用于測試Async注解的基礎(chǔ)使用) public class AsyncTestController { Autowired private AsyncService asyncService; /** * 測試基礎(chǔ)異步方法無返回值 * param taskName 任務(wù)名稱 * return 接口響應(yīng) */ GetMapping(/basic) Operation(summary 基礎(chǔ)異步方法測試, description 調(diào)用無返回值的異步方法驗證異步執(zhí)行效果) public String testBasicAsync(RequestParam String taskName) { log.info(【主線程】開始調(diào)用異步方法當(dāng)前線程{}, Thread.currentThread().getName()); // 調(diào)用異步方法 asyncService.basicAsyncTask(taskName); log.info(【主線程】異步方法調(diào)用完成無需等待結(jié)果直接返回響應(yīng)); return 異步任務(wù)已觸發(fā)可查看日志確認(rèn)執(zhí)行情況; } /** * 測試有返回值的異步方法 * param taskName 任務(wù)名稱 * param sleepTime 模擬耗時時間毫秒 * return 任務(wù)執(zhí)行結(jié)果 * throws Exception 異常 */ GetMapping(/with-return) Operation(summary 有返回值異步方法測試, description 調(diào)用有返回值的異步方法通過Future獲取執(zhí)行結(jié)果) public String testAsyncWithReturn(RequestParam String taskName, RequestParam long sleepTime) throws Exception { log.info(【主線程】開始調(diào)用有返回值的異步方法當(dāng)前線程{}, Thread.currentThread().getName()); // 調(diào)用異步方法獲取Future對象 FutureString future asyncService.asyncTaskWithReturn(taskName, sleepTime); log.info(【主線程】異步方法調(diào)用完成可繼續(xù)執(zhí)行其他邏輯); // 模擬主線程其他業(yè)務(wù)操作 log.info(【主線程】執(zhí)行其他業(yè)務(wù)邏輯...); Thread.sleep(1000); // 通過Future.get()獲取異步任務(wù)結(jié)果會阻塞直到任務(wù)完成 log.info(【主線程】開始獲取異步任務(wù)結(jié)果); String result future.get(); log.info(【主線程】異步任務(wù)結(jié)果{}, result); return 異步任務(wù)執(zhí)行結(jié)果 result; } }2.3.3 測試結(jié)果與分析測試無返回值異步方法/async/basic?taskName測試任務(wù)1 日志輸出如下關(guān)鍵觀察線程名稱和執(zhí)行順序【主線程】開始調(diào)用異步方法當(dāng)前線程http-nio-8080-exec-1 【主線程】異步方法調(diào)用完成無需等待結(jié)果直接返回響應(yīng) 【異步任務(wù)】測試任務(wù)1 開始執(zhí)行當(dāng)前線程SimpleAsyncTaskExecutor-1 【異步任務(wù)】測試任務(wù)1 執(zhí)行完畢結(jié)論主線程http-nio-8080-exec-1調(diào)用異步方法后無需等待異步任務(wù)完成直接返回響應(yīng)異步任務(wù)在獨立線程SimpleAsyncTaskExecutor-1中執(zhí)行。測試有返回值異步方法/async/with-return?taskName測試任務(wù)2sleepTime3000 日志輸出如下【主線程】開始調(diào)用有返回值的異步方法當(dāng)前線程http-nio-8080-exec-2 【主線程】異步方法調(diào)用完成可繼續(xù)執(zhí)行其他邏輯 【主線程】執(zhí)行其他業(yè)務(wù)邏輯... 【異步任務(wù)有返回值】測試任務(wù)2 開始執(zhí)行當(dāng)前線程SimpleAsyncTaskExecutor-2預(yù)計耗時3000ms 【主線程】開始獲取異步任務(wù)結(jié)果 【異步任務(wù)有返回值】測試任務(wù)2 執(zhí)行完畢 【主線程】異步任務(wù)結(jié)果測試任務(wù)2 執(zhí)行成功結(jié)論主線程調(diào)用異步方法后先執(zhí)行自身業(yè)務(wù)邏輯直到調(diào)用future.get()才會阻塞等待異步任務(wù)完成異步任務(wù)在獨立線程中執(zhí)行。三、Async進階使用自定義線程池3.1 為什么需要自定義線程池默認(rèn)情況下Async使用的是Spring提供的SimpleAsyncTaskExecutor該線程池的核心問題的是每次執(zhí)行異步任務(wù)都會創(chuàng)建一個新線程不會復(fù)用線程當(dāng)異步任務(wù)量較大時會導(dǎo)致系統(tǒng)創(chuàng)建大量線程引發(fā)線程上下文切換頻繁、內(nèi)存占用過高甚至OOM問題。因此在生產(chǎn)環(huán)境中必須自定義線程池統(tǒng)一管理線程的創(chuàng)建、復(fù)用、銷毀合理配置線程池參數(shù)。3.2 自定義線程池的3種方式3.2.1 方式1通過ConfigurationBean創(chuàng)建ThreadPoolTaskExecutor這是最常用的方式通過配置類創(chuàng)建ThreadPoolTaskExecutorSpring封裝的線程池基于JDK的ThreadPoolExecutor并指定線程池參數(shù)。package com.jam.demo.config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.util.concurrent.Executor; import java.util.concurrent.ThreadPoolExecutor; /** * 自定義線程池配置類 * author ken */ Configuration public class AsyncThreadPoolConfig { /** * 自定義線程池asyncTaskExecutor * 核心參數(shù)說明 * 1. corePoolSize核心線程數(shù)默認(rèn)活躍的線程數(shù) * 2. maxPoolSize最大線程數(shù)線程池可創(chuàng)建的最大線程數(shù) * 3. queueCapacity隊列容量核心線程滿后任務(wù)放入隊列等待 * 4. keepAliveSeconds非核心線程空閑存活時間超過該時間則銷毀 * 5. threadNamePrefix線程名稱前綴便于日志排查 * 6. rejectedExecutionHandler拒絕策略任務(wù)過多時的處理方式 * return 自定義線程池 */ Bean(name asyncTaskExecutor) public Executor asyncTaskExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); // 核心線程數(shù)根據(jù)CPU核心數(shù)配置一般為CPU核心數(shù) * 2 1 executor.setCorePoolSize(5); // 最大線程數(shù) executor.setMaxPoolSize(10); // 隊列容量核心線程滿后任務(wù)放入隊列隊列滿后才會創(chuàng)建非核心線程 executor.setQueueCapacity(25); // 非核心線程空閑存活時間30秒 executor.setKeepAliveSeconds(30); // 線程名稱前綴 executor.setThreadNamePrefix(AsyncTask-); // 拒絕策略當(dāng)線程池、隊列都滿時直接拋出異常生產(chǎn)環(huán)境可根據(jù)需求調(diào)整 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); // 初始化線程池必須調(diào)用否則線程池?zé)o法生效 executor.initialize(); return executor; } }3.2.2 方式2實現(xiàn)AsyncConfigurer接口通過實現(xiàn)AsyncConfigurer接口重寫getAsyncExecutor()方法返回自定義線程池同時可重寫getAsyncUncaughtExceptionHandler()方法自定義異步任務(wù)異常處理器。package com.jam.demo.config; import lombok.extern.slf4j.Slf4j; import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.AsyncConfigurer; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.util.concurrent.Executor; import java.util.concurrent.ThreadPoolExecutor; /** * 實現(xiàn)AsyncConfigurer接口自定義線程池 * author ken */ Configuration Slf4j public class AsyncConfigurerPoolConfig implements AsyncConfigurer { Override public Executor getAsyncExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); executor.setCorePoolSize(5); executor.setMaxPoolSize(10); executor.setQueueCapacity(25); executor.setKeepAliveSeconds(30); executor.setThreadNamePrefix(AsyncConfigurerTask-); // 拒絕策略丟棄最老的任務(wù)執(zhí)行新任務(wù) executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy()); executor.initialize(); return executor; } /** * 自定義異步任務(wù)異常處理器處理異步方法中未捕獲的異常 * return 異常處理器 */ Override public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { return (ex, method, params) - { log.error(【異步任務(wù)異常】方法{}參數(shù){}異常信息{}, method.getName(), params, ex.getMessage(), ex); }; } }3.2.3 方式3使用Async的value屬性指定線程池當(dāng)系統(tǒng)中有多個線程池時可通過Async(線程池bean名稱)指定具體使用哪個線程池。package com.jam.demo.service.impl; import com.jam.demo.service.AsyncService; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import java.util.concurrent.Future; import java.util.concurrent.FutureTask; /** * 多線程池場景下的異步服務(wù)實現(xiàn) * author ken */ Service Slf4j public class MultiPoolAsyncServiceImpl implements AsyncService { /** * 使用指定線程池asyncTaskExecutor * param taskName 任務(wù)名稱 */ Override Async(asyncTaskExecutor) public void basicAsyncTask(String taskName) { log.info(【異步任務(wù)指定線程池】{} 開始執(zhí)行當(dāng)前線程{}, taskName, Thread.currentThread().getName()); try { Thread.sleep(2000); } catch (InterruptedException e) { log.error(【異步任務(wù)指定線程池】{} 執(zhí)行異常, taskName, e); Thread.currentThread().interrupt(); } log.info(【異步任務(wù)指定線程池】{} 執(zhí)行完畢, taskName); } /** * 使用默認(rèn)線程池AsyncConfigurer配置的線程池 * param taskName 任務(wù)名稱 * param sleepTime 模擬耗時時間毫秒 * return 任務(wù)執(zhí)行結(jié)果 */ Override Async public FutureString asyncTaskWithReturn(String taskName, long sleepTime) { log.info(【異步任務(wù)默認(rèn)線程池】{} 開始執(zhí)行當(dāng)前線程{}預(yù)計耗時{}ms, taskName, Thread.currentThread().getName(), sleepTime); try { Thread.sleep(sleepTime); String result taskName 執(zhí)行成功默認(rèn)線程池; return new FutureTask(() - result); } catch (InterruptedException e) { log.error(【異步任務(wù)默認(rèn)線程池】{} 執(zhí)行異常, taskName, e); Thread.currentThread().interrupt(); return new FutureTask(() - taskName 執(zhí)行失敗默認(rèn)線程池); } } }3.3 線程池參數(shù)配置最佳實踐線程池參數(shù)的配置直接影響系統(tǒng)性能需根據(jù)業(yè)務(wù)場景合理調(diào)整核心配置原則如下參數(shù)配置建議適用場景corePoolSizeCPU核心數(shù) * 2 1CPU密集型CPU核心數(shù) * 10IO密集型CPU密集型計算任務(wù)IO密集型數(shù)據(jù)庫查詢、文件操作maxPoolSize不超過CPU核心數(shù) * 20避免線程過多導(dǎo)致上下文切換頻繁高并發(fā)場景可適當(dāng)增大queueCapacity核心線程數(shù) * 5 ~ 核心線程數(shù) * 10避免隊列過大導(dǎo)致任務(wù)堆積任務(wù)執(zhí)行時間短、數(shù)量多的場景keepAliveSeconds30 ~ 60秒非核心線程空閑時及時銷毀節(jié)省資源大多數(shù)場景通用拒絕策略核心業(yè)務(wù)AbortPolicy拋出異常便于監(jiān)控非核心業(yè)務(wù)DiscardOldestPolicy/DiscardPolicy核心業(yè)務(wù)需保證任務(wù)不丟失非核心業(yè)務(wù)可丟棄舊任務(wù)四、Async底層原理從注解解析到動態(tài)代理要真正掌握Async必須理解其底層實現(xiàn)原理。Async基于Spring AOP機制通過動態(tài)代理為目標(biāo)方法創(chuàng)建代理對象將異步調(diào)用邏輯織入代理方法中。4.1 核心執(zhí)行流程4.2 關(guān)鍵組件解析EnableAsync開啟異步支持核心是導(dǎo)入AsyncConfigurationSelector該類會根據(jù)Spring版本選擇對應(yīng)的異步配置類如ProxyAsyncConfiguration。AsyncAnnotationBeanPostProcessor后置處理器用于掃描帶有Async注解的方法為目標(biāo)類創(chuàng)建動態(tài)代理JDK動態(tài)代理或CGLIB代理。AsyncTaskExecutor異步任務(wù)執(zhí)行器即線程池是異步調(diào)用的核心載體。AnnotationAsyncExecutionInterceptor異步方法攔截器代理對象調(diào)用方法時會被該攔截器攔截負(fù)責(zé)將任務(wù)提交到線程池。4.3 動態(tài)代理機制詳解當(dāng)我們調(diào)用被Async注解的方法時實際調(diào)用的是代理對象的方法而非目標(biāo)對象的原始方法。代理對象的核心邏輯如下攔截目標(biāo)方法調(diào)用解析Async注解的屬性如指定的線程池名稱獲取對應(yīng)的線程池將目標(biāo)方法的執(zhí)行邏輯封裝為Callable或Runnable任務(wù)將任務(wù)提交到線程池由線程池中的線程執(zhí)行主線程直接返回?zé)o返回值或返回Future對象有返回值。4.4 源碼片段解析關(guān)鍵邏輯以下是AnnotationAsyncExecutionInterceptor中攔截方法的核心源碼簡化版清晰展示了異步任務(wù)的提交過程Override public Object invoke(final MethodInvocation invocation) throws Throwable { // 1. 獲取目標(biāo)方法 Class? targetClass invocation.getThis() ! null ? AopUtils.getTargetClass(invocation.getThis()) : null; Method specificMethod ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass); // 2. 解析Async注解獲取線程池 AsyncTaskExecutor executor determineAsyncExecutor(specificMethod); if (executor null) { throw new IllegalStateException(No executor specified and no default executor set); } // 3. 將目標(biāo)方法封裝為Callable任務(wù) CallableObject task () - { try { // 執(zhí)行目標(biāo)方法 Object result invocation.proceed(); if (result instanceof Future) { return ((Future?) result).get(); } } catch (Throwable ex) { // 處理異常 handleError(ex, specificMethod, invocation.getArguments()); } return null; }; // 4. 提交任務(wù)到線程池返回Future對象 return doSubmit(task, executor, specificMethod); }五、Async實戰(zhàn)進階事務(wù)處理、異常處理與批量異步5.1 異步方法與事務(wù)的關(guān)系核心結(jié)論Async注解的方法與事務(wù)注解Transactional同時使用時事務(wù)不會生效。原因如下事務(wù)基于Spring AOP需要通過代理對象調(diào)用才能生效Async的動態(tài)代理會將方法提交到線程池執(zhí)行此時目標(biāo)方法的調(diào)用脫離了事務(wù)代理的上下文事務(wù)注解無法被識別。5.1.1 錯誤示例事務(wù)不生效package com.jam.demo.service.impl; import com.jam.demo.entity.User; import com.jam.demo.mapper.UserMapper; import com.jam.demo.service.UserService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; /** * 錯誤示例異步方法事務(wù)事務(wù)不生效 * author ken */ Service Slf4j public class UserServiceImpl implements UserService { Autowired private UserMapper userMapper; /** * 錯誤示例Async與Transactional同時使用事務(wù)不生效 * 原因異步方法在獨立線程執(zhí)行脫離了事務(wù)代理上下文 * param user 用戶信息 */ Override Async Transactional(rollbackFor Exception.class) public void asyncSaveUser(User user) { userMapper.insert(user); // 模擬異常 int i 1 / 0; } }5.1.2 正確示例事務(wù)生效的異步方案要實現(xiàn)異步方法的事務(wù)控制需將事務(wù)邏輯抽離到獨立的服務(wù)方法中由異步方法調(diào)用該事務(wù)方法確保事務(wù)方法被代理對象調(diào)用。package com.jam.demo.service.impl; import com.jam.demo.entity.User; import com.jam.demo.mapper.UserMapper; import com.jam.demo.service.UserService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; /** * 正確示例異步方法事務(wù)事務(wù)生效 * 方案將事務(wù)邏輯抽離到獨立方法異步方法調(diào)用事務(wù)方法 * author ken */ Service Slf4j public class UserServiceImpl implements UserService { Autowired private UserMapper userMapper; /** * 異步方法僅負(fù)責(zé)觸發(fā)異步執(zhí)行不包含事務(wù)邏輯 * param user 用戶信息 */ Override Async(asyncTaskExecutor) public void asyncSaveUser(User user) { log.info(【異步任務(wù)】開始執(zhí)行用戶保存當(dāng)前線程{}, Thread.currentThread().getName()); // 調(diào)用事務(wù)方法 doSaveUser(user); log.info(【異步任務(wù)】用戶保存執(zhí)行完畢); } /** * 事務(wù)方法獨立的事務(wù)邏輯由Spring代理對象調(diào)用 * param user 用戶信息 */ Transactional(rollbackFor Exception.class) public void doSaveUser(User user) { userMapper.insert(user); // 模擬異常事務(wù)會回滾 int i 1 / 0; } }5.2 異步方法的異常處理異步方法中如果發(fā)生未捕獲的異常由于線程是獨立的主線程無法感知會導(dǎo)致異常丟失。因此必須配置異常處理機制。5.2.1 方式1實現(xiàn)AsyncUncaughtExceptionHandler全局異常處理如3.2.2節(jié)所示通過實現(xiàn)AsyncConfigurer接口的getAsyncUncaughtExceptionHandler()方法配置全局異步異常處理器處理所有無返回值異步方法的未捕獲異常。5.2.2 方式2通過Future獲取異常有返回值方法有返回值的異步方法會返回Future對象調(diào)用Future.get()方法時會將異步方法中的異常拋出可通過try-catch捕獲。// 示例捕獲有返回值異步方法的異常 GetMapping(/with-return-exception) Operation(summary 有返回值異步方法異常處理測試) public String testAsyncWithReturnException() { log.info(【主線程】開始調(diào)用有返回值的異步方法); FutureString future asyncService.asyncTaskWithReturn(測試異常任務(wù), 2000); try { String result future.get(); return result; } catch (Exception e) { log.error(【主線程】捕獲異步任務(wù)異常, e); return 異步任務(wù)執(zhí)行失敗 e.getMessage(); } }5.2.3 方式3自定義異常處理器局部異常處理通過Async的exceptionHandler屬性指定局部異常處理器需Spring 4.1版本支持。package com.jam.demo.handler; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; /** * 自定義異步異常處理器局部 * author ken */ Component Slf4j public class CustomAsyncExceptionHandler { /** * 處理異步方法異常 * param ex 異常對象 * param method 方法對象 * param params 方法參數(shù) */ public void handleException(Throwable ex, String method, Object... params) { log.error(【自定義異步異常】方法{}參數(shù){}異常信息{}, method, params, ex.getMessage(), ex); } }// 使用局部異常處理器 Async(exceptionHandler customAsyncExceptionHandler) public void asyncTaskWithCustomExceptionHandler(String taskName) { log.info(【異步任務(wù)自定義異常處理器】{} 開始執(zhí)行, taskName); // 模擬異常 int i 1 / 0; }5.3 批量異步任務(wù)處理在實際開發(fā)中經(jīng)常需要批量執(zhí)行異步任務(wù)如批量發(fā)送短信、批量處理數(shù)據(jù)此時可通過CompletableFuture實現(xiàn)批量任務(wù)的并發(fā)執(zhí)行、結(jié)果聚合、異常捕獲。5.3.1 批量異步任務(wù)示例基于CompletableFuturepackage com.jam.demo.service.impl; import com.jam.demo.service.BatchAsyncService; import com.google.common.collect.Lists; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import java.util.List; import java.util.concurrent.CompletableFuture; /** * 批量異步任務(wù)服務(wù)實現(xiàn) * author ken */ Service Slf4j public class BatchAsyncServiceImpl implements BatchAsyncService { /** * 單個批量任務(wù)的異步方法 * param taskId 任務(wù)ID * return CompletableFutureString 任務(wù)執(zhí)行結(jié)果 */ Async(asyncTaskExecutor) public CompletableFutureString batchTask(Integer taskId) { log.info(【批量異步任務(wù)】任務(wù){(diào)} 開始執(zhí)行當(dāng)前線程{}, taskId, Thread.currentThread().getName()); try { // 模擬耗時操作如處理單條數(shù)據(jù) Thread.sleep(1000); String result 任務(wù) taskId 執(zhí)行成功; return CompletableFuture.completedFuture(result); } catch (InterruptedException e) { log.error(【批量異步任務(wù)】任務(wù){(diào)} 執(zhí)行異常, taskId, e); Thread.currentThread().interrupt(); return CompletableFuture.failedFuture(e); } } /** * 批量執(zhí)行異步任務(wù)聚合結(jié)果 * param taskCount 任務(wù)數(shù)量 * return 所有任務(wù)的執(zhí)行結(jié)果 */ Override public CompletableFutureListString executeBatchTasks(Integer taskCount) { // 生成任務(wù)列表 ListInteger taskIds Lists.newArrayList(); for (int i 1; i taskCount; i) { taskIds.add(i); } // 批量提交異步任務(wù)獲取CompletableFuture列表 ListCompletableFutureString futureList taskIds.stream() .map(this::batchTask) .toList(); // 聚合所有任務(wù)結(jié)果當(dāng)所有任務(wù)完成后收集結(jié)果 return CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])) .thenApply(v - futureList.stream() .map(CompletableFuture::join) .toList()); } }5.3.2 測試批量異步任務(wù)GetMapping(/batch) Operation(summary 批量異步任務(wù)測試, description 批量執(zhí)行異步任務(wù)聚合所有任務(wù)結(jié)果) public CompletableFutureString testBatchAsyncTasks(RequestParam Integer taskCount) { log.info(【主線程】開始批量提交異步任務(wù)任務(wù)數(shù)量{}, taskCount); CompletableFutureListString batchFuture batchAsyncService.executeBatchTasks(taskCount); return batchFuture.thenApply(results - { log.info(【主線程】所有批量異步任務(wù)執(zhí)行完畢結(jié)果{}, results); return 批量任務(wù)執(zhí)行完成共 results.size() 個任務(wù)結(jié)果 results; }); }六、Async常見坑與避坑指南6.1 坑1異步方法不生效6.1.1 常見原因未添加EnableAsync注解異步方法為private修飾Spring AOP無法攔截private方法異步方法被同一個類中的其他方法調(diào)用內(nèi)部調(diào)用未經(jīng)過代理對象自定義線程池未調(diào)用initialize()方法線程池未初始化依賴注入的是目標(biāo)對象而非代理對象如使用new關(guān)鍵字創(chuàng)建對象。6.1.2 避坑方案確保啟動類或配置類上添加EnableAsync異步方法必須為public修飾避免內(nèi)部調(diào)用異步方法和調(diào)用方必須在不同的類中自定義線程池時必須調(diào)用executor.initialize()依賴注入使用Autowired或Resource避免使用new關(guān)鍵字創(chuàng)建對象。6.1.3 錯誤示例與正確示例// 錯誤示例1內(nèi)部調(diào)用異步不生效 Service public class AsyncErrorService { // 內(nèi)部調(diào)用異步方法未經(jīng)過代理對象 public void callAsyncMethod() { asyncMethod(); } Async public void asyncMethod() { // 異步邏輯 } } // 錯誤示例2private修飾異步不生效 Service public class AsyncErrorService { Async private void asyncMethod() { // 異步邏輯 } } // 正確示例不同類調(diào)用異步生效 Service public class AsyncCallerService { Autowired private AsyncService asyncService; // 調(diào)用不同類中的異步方法經(jīng)過代理對象 public void callAsyncMethod() { asyncService.basicAsyncTask(正確示例任務(wù)); } }6.2 坑2線程池耗盡6.2.1 常見原因使用默認(rèn)線程池SimpleAsyncTaskExecutor每次創(chuàng)建新線程無上限線程池參數(shù)配置不合理核心線程數(shù)、最大線程數(shù)過小隊列容量過小異步任務(wù)執(zhí)行時間過長導(dǎo)致線程被長時間占用任務(wù)提交速度超過線程池處理速度導(dǎo)致任務(wù)堆積、線程池滿負(fù)荷。6.2.2 避坑方案生產(chǎn)環(huán)境禁用默認(rèn)線程池必須自定義線程池根據(jù)業(yè)務(wù)場景合理配置線程池參數(shù)參考3.3節(jié)最佳實踐監(jiān)控異步任務(wù)執(zhí)行時間優(yōu)化耗時任務(wù)如拆分大任務(wù)、優(yōu)化SQL配置合理的拒絕策略避免任務(wù)過多時系統(tǒng)崩潰對異步任務(wù)進行限流避免短時間內(nèi)提交大量任務(wù)。6.3 坑3事務(wù)不生效6.3.1 常見原因如5.1節(jié)所述Async與Transactional同時使用時異步方法脫離了事務(wù)代理上下文導(dǎo)致事務(wù)不生效。6.3.2 避坑方案將事務(wù)邏輯抽離到獨立的public方法中由異步方法調(diào)用該事務(wù)方法確保事務(wù)方法被代理對象調(diào)用具體示例參考5.1.2節(jié)。6.4 坑4異常丟失6.4.1 常見原因無返回值的異步方法中發(fā)生未捕獲的異常由于線程獨立主線程無法感知導(dǎo)致異常丟失。6.4.2 避坑方案配置全局或局部異常處理器參考5.2節(jié)確保所有異步方法的異常都能被捕獲和記錄。6.5 坑5異步方法返回值錯誤6.5.1 常見原因有返回值的異步方法未返回Future或其實現(xiàn)類如直接返回String、Integer等基本類型導(dǎo)致無法獲取異步執(zhí)行結(jié)果。6.5.2 避坑方案有返回值的異步方法必須返回Future或其實現(xiàn)類如FutureTask、CompletableFuture通過Future.get()獲取執(zhí)行結(jié)果。// 錯誤示例返回值不是Future無法獲取異步結(jié)果 Async public String asyncTaskWithWrongReturn(String taskName) { return taskName 執(zhí)行成功; } // 正確示例返回Future可獲取異步結(jié)果 Async public FutureString asyncTaskWithCorrectReturn(String taskName) { return new FutureTask(() - taskName 執(zhí)行成功); }七、Async實戰(zhàn)案例用戶注冊異步通知系統(tǒng)7.1 案例需求用戶注冊成功后需要執(zhí)行以下3個異步任務(wù)發(fā)送注冊成功短信發(fā)送注冊成功郵件記錄用戶注冊日志到數(shù)據(jù)庫。要求3個任務(wù)并發(fā)執(zhí)行提升注冊接口響應(yīng)速度確保每個任務(wù)的異常都能被捕獲記錄日志的任務(wù)需要事務(wù)支持確保日志數(shù)據(jù)入庫成功。7.2 案例實現(xiàn)7.2.1 數(shù)據(jù)庫表設(shè)計MySQL-- 用戶表 CREATE TABLE user ( id bigint NOT NULL AUTO_INCREMENT COMMENT 主鍵ID, username varchar(50) NOT NULL COMMENT 用戶名, phone varchar(20) NOT NULL COMMENT 手機號, email varchar(100) NOT NULL COMMENT 郵箱, create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 創(chuàng)建時間, PRIMARY KEY (id), UNIQUE KEY uk_username (username) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT用戶表; -- 注冊日志表 CREATE TABLE user_register_log ( id bigint NOT NULL AUTO_INCREMENT COMMENT 主鍵ID, user_id bigint NOT NULL COMMENT 用戶ID, register_ip varchar(50) NOT NULL COMMENT 注冊IP, log_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 日志時間, PRIMARY KEY (id), KEY idx_user_id (user_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT用戶注冊日志表;7.2.2 實體類package com.jam.demo.entity; import com.baomidou.mybatisplus.annotation.IdType; import com.baomidou.mybatisplus.annotation.TableId; import com.baomidou.mybatisplus.annotation.TableName; import lombok.Data; import java.time.LocalDateTime; /** * 用戶實體類 * author ken */ Data TableName(user) public class User { /** * 主鍵ID */ TableId(type IdType.AUTO) private Long id; /** * 用戶名 */ private String username; /** * 手機號 */ private String phone; /** * 郵箱 */ private String email; /** * 創(chuàng)建時間 */ private LocalDateTime createTime; }package com.jam.demo.entity; import com.baomidou.mybatisplus.annotation.IdType; import com.baomidou.mybatisplus.annotation.TableId; import com.baomidou.mybatisplus.annotation.TableName; import lombok.Data; import java.time.LocalDateTime; /** * 用戶注冊日志實體類 * author ken */ Data TableName(user_register_log) public class UserRegisterLog { /** * 主鍵ID */ TableId(type IdType.AUTO) private Long id; /** * 用戶ID */ private Long userId; /** * 注冊IP */ private String registerIp; /** * 日志時間 */ private LocalDateTime logTime; }7.2.3 Mapper層package com.jam.demo.mapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.jam.demo.entity.User; import org.springframework.stereotype.Repository; /** * 用戶Mapper * author ken */ Repository public interface UserMapper extends BaseMapperUser { }package com.jam.demo.mapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.jam.demo.entity.UserRegisterLog; import org.springframework.stereotype.Repository; /** * 注冊日志Mapper * author ken */ Repository public interface UserRegisterLogMapper extends BaseMapperUserRegisterLog { }7.2.4 服務(wù)層package com.jam.demo.service; import com.jam.demo.entity.User; import com.jam.demo.entity.UserRegisterLog; /** * 注冊相關(guān)服務(wù) * author ken */ public interface RegisterService { /** * 用戶注冊同步方法保存用戶信息 * param user 用戶信息 * param registerIp 注冊IP * return 注冊成功的用戶ID */ Long userRegister(User user, String registerIp); /** * 發(fā)送注冊成功短信異步方法 * param phone 手機號 */ void sendRegisterSms(String phone); /** * 發(fā)送注冊成功郵件異步方法 * param email 郵箱 */ void sendRegisterEmail(String email); /** * 記錄注冊日志異步事務(wù)方法 * param userId 用戶ID * param registerIp 注冊IP */ void recordRegisterLog(Long userId, String registerIp); }package com.jam.demo.service.impl; import com.jam.demo.entity.User; import com.jam.demo.entity.UserRegisterLog; import com.jam.demo.mapper.UserMapper; import com.jam.demo.mapper.UserRegisterLogMapper; import com.jam.demo.service.RegisterService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import java.time.LocalDateTime; /** * 注冊相關(guān)服務(wù)實現(xiàn) * author ken */ Service Slf4j public class RegisterServiceImpl implements RegisterService { Autowired private UserMapper userMapper; Autowired private UserRegisterLogMapper userRegisterLogMapper; /** * 用戶注冊同步方法保存用戶信息 * 注用戶注冊是核心流程需同步執(zhí)行確保用戶信息入庫成功后再觸發(fā)異步任務(wù) * param user 用戶信息 * param registerIp 注冊IP * return 注冊成功的用戶ID */ Override public Long userRegister(User user, String registerIp) { log.info(【用戶注冊】開始保存用戶信息用戶名{}, user.getUsername()); // 保存用戶信息 user.setCreateTime(LocalDateTime.now()); userMapper.insert(user); Long userId user.getId(); log.info(【用戶注冊】用戶信息保存成功用戶ID{}, userId); // 觸發(fā)3個異步任務(wù)并發(fā)執(zhí)行 sendRegisterSms(user.getPhone()); sendRegisterEmail(user.getEmail()); recordRegisterLog(userId, registerIp); return userId; } /** * 發(fā)送注冊成功短信異步方法 * param phone 手機號 */ Override Async(asyncTaskExecutor) public void sendRegisterSms(String phone) { log.info(【異步任務(wù)-發(fā)送短信】開始向手機號{} 發(fā)送注冊成功短信當(dāng)前線程{}, phone, Thread.currentThread().getName()); // 模擬短信發(fā)送耗時 try { Thread.sleep(1500); } catch (InterruptedException e) { log.error(【異步任務(wù)-發(fā)送短信】向手機號{} 發(fā)送短信異常, phone, e); Thread.currentThread().interrupt(); } log.info(【異步任務(wù)-發(fā)送短信】向手機號{} 發(fā)送短信成功, phone); } /** * 發(fā)送注冊成功郵件異步方法 * param email 郵箱 */ Override Async(asyncTaskExecutor) public void sendRegisterEmail(String email) { log.info(【異步任務(wù)-發(fā)送郵件】開始向郵箱{} 發(fā)送注冊成功郵件當(dāng)前線程{}, email, Thread.currentThread().getName()); // 模擬郵件發(fā)送耗時 try { Thread.sleep(2000); } catch (InterruptedException e) { log.error(【異步任務(wù)-發(fā)送郵件】向郵箱{} 發(fā)送郵件異常, email, e); Thread.currentThread().interrupt(); } log.info(【異步任務(wù)-發(fā)送郵件】向郵箱{} 發(fā)送郵件成功, email); } /** * 記錄注冊日志異步事務(wù)方法 * 注事務(wù)邏輯抽離到獨立方法確保事務(wù)生效 * param userId 用戶ID * param registerIp 注冊IP */ Override Async(asyncTaskExecutor) public void recordRegisterLog(Long userId, String registerIp) { log.info(【異步任務(wù)-記錄日志】開始記錄用戶{} 的注冊日志當(dāng)前線程{}, userId, Thread.currentThread().getName()); doRecordRegisterLog(userId, registerIp); log.info(【異步任務(wù)-記錄日志】用戶{} 的注冊日志記錄成功, userId); } /** * 事務(wù)方法實際執(zhí)行日志記錄邏輯 * param userId 用戶ID * param registerIp 注冊IP */ Transactional(rollbackFor Exception.class) public void doRecordRegisterLog(Long userId, String registerIp) { UserRegisterLog log new UserRegisterLog(); log.setUserId(userId); log.setRegisterIp(registerIp); log.setLogTime(LocalDateTime.now()); userRegisterLogMapper.insert(log); // 模擬異常測試事務(wù)回滾 // int i 1 / 0; } }7.2.5 控制器層package com.jam.demo.controller; import com.jam.demo.entity.User; import com.jam.demo.service.RegisterService; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.tags.Tag; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import org.springframework.util.StringUtils; /** * 用戶注冊控制器 * author ken */ RestController RequestMapping(/register) Slf4j Tag(name 用戶注冊接口, description 用戶注冊核心接口包含異步通知功能) public class RegisterController { Autowired private RegisterService registerService; /** * 用戶注冊接口 * param user 用戶信息 * param registerIp 注冊IP * return 注冊結(jié)果 */ PostMapping Operation(summary 用戶注冊, description 用戶注冊成功后異步發(fā)送短信、郵件記錄注冊日志) public String userRegister(RequestBody User user, RequestParam String registerIp) { // 參數(shù)校驗 if (ObjectUtils.isEmpty(user) || !StringUtils.hasText(user.getUsername()) || !StringUtils.hasText(user.getPhone()) || !StringUtils.hasText(user.getEmail())) { return 參數(shù)錯誤用戶名、手機號、郵箱不能為空; } if (!StringUtils.hasText(registerIp)) { return 參數(shù)錯誤注冊IP不能為空; } log.info(【用戶注冊接口】開始處理注冊請求用戶名{}, user.getUsername()); // 執(zhí)行注冊同步觸發(fā)異步任務(wù) Long userId registerService.userRegister(user, registerIp); log.info(【用戶注冊接口】注冊請求處理完成用戶ID{}已觸發(fā)異步通知任務(wù), userId); return 注冊成功用戶ID userId; } }7.3 案例測試與結(jié)果分析發(fā)送POST請求到/register請求參數(shù)如下{ username: test_user, phone: 13800138000, email: testexample.com }請求參數(shù)registerIp127.0.0.1日志輸出如下關(guān)鍵觀察線程名稱和執(zhí)行順序【用戶注冊接口】開始處理注冊請求用戶名test_user 【用戶注冊】開始保存用戶信息用戶名test_user 【用戶注冊】用戶信息保存成功用戶ID1 【異步任務(wù)-發(fā)送短信】開始向手機號13800138000 發(fā)送注冊成功短信當(dāng)前線程AsyncTask-1 【異步任務(wù)-發(fā)送郵件】開始向郵箱testexample.com 發(fā)送注冊成功郵件當(dāng)前線程AsyncTask-2 【異步任務(wù)-記錄日志】開始記錄用戶1 的注冊日志當(dāng)前線程AsyncTask-3 【用戶注冊接口】注冊請求處理完成用戶ID1已觸發(fā)異步通知任務(wù) 【異步任務(wù)-發(fā)送短信】向手機號13800138000 發(fā)送短信成功 【異步任務(wù)-記錄日志】用戶1 的注冊日志記錄成功 【異步任務(wù)-發(fā)送郵件】向郵箱testexample.com 發(fā)送郵件成功結(jié)果分析核心流程同步執(zhí)行用戶信息保存是核心流程同步執(zhí)行確保用戶注冊成功后才觸發(fā)后續(xù)異步任務(wù)異步任務(wù)并發(fā)執(zhí)行發(fā)送短信、發(fā)送郵件、記錄日志3個任務(wù)在3個獨立線程AsyncTask-1、AsyncTask-2、AsyncTask-3中并發(fā)執(zhí)行無需順序等待接口響應(yīng)快速主線程用戶注冊接口在觸發(fā)異步任務(wù)后立即返回響應(yīng)無需等待異步任務(wù)完成提升了接口響應(yīng)速度異常隔離每個異步任務(wù)的異常都被獨立捕獲不會影響其他任務(wù)和主線程的執(zhí)行。八、Async監(jiān)控與調(diào)優(yōu)8.1 異步任務(wù)監(jiān)控在生產(chǎn)環(huán)境中需要對異步任務(wù)的執(zhí)行狀態(tài)、線程池狀態(tài)進行監(jiān)控以便及時發(fā)現(xiàn)問題。常用監(jiān)控方案如下8.1.1 基于Spring Boot Actuator監(jiān)控線程池Spring Boot Actuator提供了線程池監(jiān)控端點可通過配置暴露線程池相關(guān)指標(biāo)。添加依賴dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-actuator/artifactId version3.2.5/version /dependency dependency groupIdio.micrometer/groupId artifactIdmicrometer-registry-prometheus/artifactId version1.12.5/version /dependency配置application.ymlspring: application: name: async-demo management: endpoints: web: exposure: include: health,info,metrics,threadpool # 暴露線程池監(jiān)控端點 metrics: tags: application: ${spring.application.name} export: prometheus: enabled: true # 啟用Prometheus導(dǎo)出指標(biāo)自定義線程池監(jiān)控指標(biāo)package com.jam.demo.config; import io.micrometer.core.instrument.Gauge; import io.micrometer.core.instrument.MeterRegistry; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import javax.annotation.PostConstruct; import java.util.concurrent.ThreadPoolExecutor; /** * 線程池監(jiān)控配置 * author ken */ Configuration public class ThreadPoolMonitorConfig { Autowired private MeterRegistry meterRegistry; Autowired Qualifier(asyncTaskExecutor) private ThreadPoolTaskExecutor asyncTaskExecutor; /** * 注冊線程池監(jiān)控指標(biāo) */ PostConstruct public void monitorThreadPool() { ThreadPoolExecutor executor asyncTaskExecutor.getThreadPoolExecutor(); // 核心線程數(shù) Gauge.builder(threadpool.core.size, executor, ThreadPoolExecutor::getCorePoolSize) .tag(threadpool.name, asyncTaskExecutor) .register(meterRegistry); // 活躍線程數(shù) Gauge.builder(threadpool.active.size, executor, ThreadPoolExecutor::getActiveCount) .tag(threadpool.name, asyncTaskExecutor) .register(meterRegistry); // 最大線程數(shù) Gauge.builder(threadpool.max.size, executor, ThreadPoolExecutor::getMaximumPoolSize) .tag(threadpool.name, asyncTaskExecutor) .register(meterRegistry); // 隊列中的任務(wù)數(shù) Gauge.builder(threadpool.queue.size, executor, e - e.getQueue().size()) .tag(threadpool.name, asyncTaskExecutor) .register(meterRegistry); // 已完成的任務(wù)數(shù) Gauge.builder(threadpool.completed.tasks, executor, ThreadPoolExecutor::getCompletedTaskCount) .tag(threadpool.name, asyncTaskExecutor) .register(meterRegistry); } }訪問監(jiān)控端點查看線程池指標(biāo)http://localhost:8080/actuator/metrics/threadpool.active.size?tagthreadpool.name:asyncTaskExecutor查看Prometheus指標(biāo)http://localhost:8080/actuator/prometheus可結(jié)合Grafana可視化展示8.1.2 自定義異步任務(wù)執(zhí)行日志通過AOP切面記錄異步任務(wù)的執(zhí)行時間、參數(shù)、結(jié)果等信息便于問題排查。package com.jam.demo.aspect; import lombok.extern.slf4j.Slf4j; import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.annotation.Around; import org.aspectj.lang.annotation.Aspect; import org.aspectj.lang.annotation.Pointcut; import org.springframework.stereotype.Component; /** * 異步任務(wù)執(zhí)行日志切面 * author ken */ Aspect Component Slf4j public class AsyncTaskLogAspect { /** * 切入點所有被Async注解的方法 */ Pointcut(annotation(org.springframework.scheduling.annotation.Async)) public void asyncTaskPointcut() {} /** * 環(huán)繞通知記錄任務(wù)執(zhí)行時間、參數(shù)、結(jié)果 * param joinPoint 連接點 * return 任務(wù)執(zhí)行結(jié)果 * throws Throwable 異常 */ Around(asyncTaskPointcut()) public Object aroundAsyncTask(ProceedingJoinPoint joinPoint) throws Throwable { // 記錄開始時間 long startTime System.currentTimeMillis(); String methodName joinPoint.getSignature().getDeclaringTypeName() . joinPoint.getSignature().getName(); Object[] args joinPoint.getArgs(); log.info(【異步任務(wù)監(jiān)控】{} 開始執(zhí)行參數(shù){}當(dāng)前線程{}, methodName, args, Thread.currentThread().getName()); try { // 執(zhí)行目標(biāo)方法 Object result joinPoint.proceed(); // 記錄執(zhí)行時間和結(jié)果 long costTime System.currentTimeMillis() - startTime; log.info(【異步任務(wù)監(jiān)控】{} 執(zhí)行完成耗時{}ms結(jié)果{}, methodName, costTime, result); return result; } catch (Throwable e) { log.error(【異步任務(wù)監(jiān)控】{} 執(zhí)行異常耗時{}ms, methodName, System.currentTimeMillis() - startTime, e); throw e; } } }8.2 異步任務(wù)調(diào)優(yōu)8.2.1 線程池參數(shù)動態(tài)調(diào)優(yōu)在生產(chǎn)環(huán)境中線程池參數(shù)可能需要根據(jù)業(yè)務(wù)流量動態(tài)調(diào)整可通過以下方案實現(xiàn)基于配置中心如Nacos、Apollo動態(tài)刷新線程池參數(shù)提供接口手動調(diào)整線程池參數(shù)需做好權(quán)限控制。示例基于Nacos動態(tài)調(diào)整線程池參數(shù)package com.jam.demo.config; import com.alibaba.nacos.api.config.annotation.NacosConfigListener; import com.alibaba.nacos.api.config.annotation.NacosValue; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import javax.annotation.Resource; /** * 基于Nacos的線程池動態(tài)配置 * author ken */ Configuration public class DynamicThreadPoolConfig { Resource Qualifier(asyncTaskExecutor) private ThreadPoolTaskExecutor asyncTaskExecutor; // 從Nacos獲取核心線程數(shù)配置 NacosValue(value ${threadpool.async.core-size:5}, autoRefreshed true) private int corePoolSize; // 從Nacos獲取最大線程數(shù)配置 NacosValue(value ${threadpool.async.max-size:10}, autoRefreshed true) private int maxPoolSize; // 從Nacos獲取空閑存活時間配置 NacosValue(value ${threadpool.async.keep-alive-seconds:30}, autoRefreshed true) private int keepAliveSeconds; /** * 監(jiān)聽配置變化動態(tài)調(diào)整線程池參數(shù) */ NacosConfigListener(dataId async-demo-threadpool-config, groupId DEFAULT_GROUP) public void refreshThreadPoolConfig(String config) { // 解析配置此處簡化實際需解析JSON/Properties格式 // 動態(tài)調(diào)整核心線程數(shù) asyncTaskExecutor.setCorePoolSize(corePoolSize); // 動態(tài)調(diào)整最大線程數(shù) asyncTaskExecutor.setMaxPoolSize(maxPoolSize); // 動態(tài)調(diào)整空閑存活時間 asyncTaskExecutor.setKeepAliveSeconds(keepAliveSeconds); // 重新初始化線程池僅調(diào)整核心線程數(shù)、最大線程數(shù)、空閑存活時間時無需重新初始化 asyncTaskExecutor.initialize(); } }8.2.2 任務(wù)拆分與合并對于執(zhí)行時間過長的大任務(wù)可拆分為多個小任務(wù)并發(fā)執(zhí)行提升執(zhí)行效率對于大量小任務(wù)可合并為批次任務(wù)執(zhí)行減少線程切換開銷。示例大任務(wù)拆分/** * 大任務(wù)拆分示例批量處理1000條數(shù)據(jù)拆分為10個小任務(wù)每個任務(wù)處理100條 * param dataList 待處理數(shù)據(jù)列表 * return 處理結(jié)果 */ Async(asyncTaskExecutor) public CompletableFutureVoid processLargeData(ListString dataList) { // 拆分任務(wù)每100條數(shù)據(jù)為一個子任務(wù) ListListString subTasks Lists.partition(dataList, 100); // 并發(fā)執(zhí)行子任務(wù) ListCompletableFutureVoid futureList subTasks.stream() .map(subList - CompletableFuture.runAsync(() - processSubTask(subList), asyncTaskExecutor)) .toList(); // 等待所有子任務(wù)完成 return CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])); } /** * 子任務(wù)處理邏輯 * param subList 子任務(wù)數(shù)據(jù)列表 */ private void processSubTask(ListString subList) { log.info(【子任務(wù)】開始處理 {} 條數(shù)據(jù)當(dāng)前線程{}, subList.size(), Thread.currentThread().getName()); // 處理邏輯 subList.forEach(data - { // 數(shù)據(jù)處理操作 }); log.info(【子任務(wù)】數(shù)據(jù)處理完成); }8.2.3 避免異步任務(wù)嵌套異步任務(wù)內(nèi)部盡量不要嵌套調(diào)用其他異步任務(wù)否則會導(dǎo)致線程池資源被過度占用增加系統(tǒng)復(fù)雜度和排查難度。若必須嵌套需嚴(yán)格控制嵌套層級和任務(wù)數(shù)量。九、總結(jié)與核心要點回顧Async注解是Spring框架中實現(xiàn)異步編程的核心工具其使用簡單但要在生產(chǎn)環(huán)境中穩(wěn)定運行需掌握以下核心要點基礎(chǔ)使用必須添加EnableAsync開啟異步支持異步方法需為public修飾避免內(nèi)部調(diào)用線程池配置生產(chǎn)環(huán)境禁用默認(rèn)線程池自定義線程池需合理配置核心參數(shù)調(diào)用initialize()初始化事務(wù)處理Async與Transactional同時使用時事務(wù)不生效需將事務(wù)邏輯抽離到獨立方法異常處理通過AsyncUncaughtExceptionHandler或Future捕獲異常避免異常丟失避坑指南重點關(guān)注異步不生效、線程池耗盡、事務(wù)失效、異常丟失等常見問題監(jiān)控調(diào)優(yōu)通過Actuator、自定義日志實現(xiàn)監(jiān)控結(jié)合配置中心實現(xiàn)動態(tài)調(diào)優(yōu)合理拆分/合并任務(wù)提升效率。通過本文的講解和實戰(zhàn)示例相信你已全面掌握Async注解的使用方法和底層邏輯。在實際開發(fā)中需結(jié)合業(yè)務(wù)場景靈活運用異步編程平衡系統(tǒng)吞吐量和穩(wěn)定性讓異步成為提升系統(tǒng)性能的利器。