Spring Boot之@Async異步線程池示例詳解
前言
很多業(yè)務場景需要使用異步去完成，比如：發(fā)送短信通知。要完成異步操作一般有兩種：
- 1、消息隊列MQ
- 2、線程池處理。
我們來看看Spring框架中如何去使用線程池來完成異步操作，以及分析背后的原理。
一. Spring異步線程池的接口類 ：TaskExecutor
在Spring4中，Spring中引入了一個新的注解@Async，這個注解讓我們在使用Spring完成異步操作變得非常方便。
Spring異步線程池的接口類，其實質是java.util.concurrent.Executor
Spring 已經(jīng)實現(xiàn)的異常線程池：
1. SimpleAsyncTaskExecutor：不是真的線程池，這個類不重用線程，每次調用都會創(chuàng)建一個新的線程。
2. SyncTaskExecutor：這個類沒有實現(xiàn)異步調用，只是一個同步操作。只適用于不需要多線程的地方
3. ConcurrentTaskExecutor：Executor的適配類，不推薦使用。如果ThreadPoolTaskExecutor不滿足要求時，才用考慮使用這個類
4. SimpleThreadPoolTaskExecutor：是Quartz的SimpleThreadPool的類。線程池同時被quartz和非quartz使用，才需要使用此類
5. ThreadPoolTaskExecutor ：最常使用，推薦。 其實質是對java.util.concurrent.ThreadPoolExecutor的包裝，
關于java-多線程和線程池：http://www.dbjr.com.cn/article/222986.htm
我們查看ThreadPoolExecutor初始化的源碼就知道使用ThreadPoolExecutor。
二、簡單使用說明
Spring中用@Async注解標記的方法，稱為異步方法。在spring boot應用中使用@Async很簡單：
1、調用異步方法類上或者啟動類加上注解@EnableAsync
2、在需要被異步調用的方法外加上@Async
3、所使用的@Async注解方法的類對象應該是Spring容器管理的bean對象；
啟動類加上注解@EnableAsync：
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;
@SpringBootApplication
@EnableAsync
public class CollectorApplication {
public static void main(String[] args) throws Exception {
SpringApplication.run(CollectorApplication.class, args);
}
}
在需要被異步調用的方法外加上@Async，同時類AsyncService加上注解@Service或者@Component，使其對象成為Spring容器管理的bean對象；
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
@Service
@Transactional
public class AsyncService {
@Async
public void asyncMethod(String s) {
System.out.println("receive:" + s);
}
public void test() {
System.out.println("test");
asyncMethod();//同一個類里面調用異步方法
}
@Async
public void test2() {
AsyncService asyncService = context.getBean(AsyncService.class);
asyncService.asyncMethod();//異步
}
/**
* 異布調用返回Future
*/
@Async
public Future<String> asyncInvokeReturnFuture(int i) {
System.out.println("asyncInvokeReturnFuture, parementer="+ i);
Future<String> future;
try {
Thread.sleep(1000 * 1);
future = new AsyncResult<String>("success:" + i);
} catch (InterruptedException e) {
future = new AsyncResult<String>("error");
}
return future;
}
}
//異步方法和普通的方法調用相同
asyncService.asyncMethod("123");
Future<String> future = asyncService.asyncInvokeReturnFuture(100);
System.out.println(future.get());
如果將一個類聲明為異步類@Async，那么這個類對外暴露的方法全部成為異步方法。
@Async
@Service
public class AsyncClass {
public AsyncClass() {
System.out.println("----init AsyncClass----");
}
volatile int index = 0;
public void foo() {
System.out.println("asyncclass foo, index:" + index);
}
public void foo(int i) {
this.index = i;
System.out.println("asyncclass foo, index:" + i);
}
public void bar(int i) {
this.index = i;
System.out.println("asyncclass bar, index:" + i);
}
}
這里需要注意的是：
1、同一個類里面調用異步方法不生效：原因默認類內的方法調用不會被aop攔截，即調用方和被調用方是在同一個類中，是無法產生切面的，該對象沒有被Spring容器管理。即@Async方法不生效。
解決辦法：如果要使同一個類中的方法之間調用也被攔截，需要使用spring容器中的實例對象，而不是使用默認的this，因為通過bean實例的調用才會被spring的aop攔截
本例使用方法：AsyncService asyncService = context.getBean(AsyncService.class); 然后使用這個引用調用本地的方法即可達到被攔截的目的
備注：這種方法只能攔截protected，default，public方法，private方法無法攔截。這個是spring aop的一個機制。
2、如果不自定義異步方法的線程池默認使用SimpleAsyncTaskExecutor。SimpleAsyncTaskExecutor：不是真的線程池，這個類不重用線程，每次調用都會創(chuàng)建一個新的線程。并發(fā)大的時候會產生嚴重的性能問題。
3、異步方法返回類型只能有兩種：void和java.util.concurrent.Future。
1）當返回類型為void的時候，方法調用過程產生的異常不會拋到調用者層面，
可以通過注AsyncUncaughtExceptionHandler來捕獲此類異常
2）當返回類型為Future的時候，方法調用過程產生的異常會拋到調用者層面
三、定義通用線程池
1、定義線程池
在Spring Boot主類中定義一個線程池，public Executor taskExecutor() 方法用于自定義自己的線程池，線程池前綴”taskExecutor-”。如果不定義，則使用系統(tǒng)默認的線程池。
@SpringBootApplication
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
@EnableAsync
@Configuration
class TaskPoolConfig {
@Bean
public Executor taskExecutor1() {
ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
pool.setCorePoolSize(5); //線程池活躍的線程數(shù)
pool.setMaxPoolSize(10); //線程池最大活躍的線程數(shù)
pool.setWaitForTasksToCompleteOnShutdown(true);
pool.setThreadNamePrefix("defaultExecutor");
return pool;
}
@Bean("taskExecutor")
public Executor taskExecutor2() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(10);
executor.setMaxPoolSize(20);
executor.setQueueCapacity(200);
executor.setKeepAliveSeconds(60);
executor.setThreadNamePrefix("taskExecutor-");
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setAwaitTerminationSeconds(60);
return executor;
}
}
}
上面我們通過ThreadPoolTaskExecutor創(chuàng)建了一個線程池，同時設置了如下參數(shù)：
- 核心線程數(shù)10：線程池創(chuàng)建時初始化的線程數(shù)
- 最大線程數(shù)20：線程池最大的線程數(shù)，只有在緩沖隊列滿了之后才會申請超過核心線程數(shù)的線程
- 緩沖隊列200：用來緩沖執(zhí)行任務的隊列
- 允許線程的空閑時間60秒：超過了核心線程數(shù)之外的線程，在空閑時間到達之后會被銷毀
- 線程池名的前綴：設置好了之后可以方便我們定位處理任務所在的線程池
- 線程池對拒絕任務的處理策略：此處采用了CallerRunsPolicy策略，當線程池沒有處理能力的時候，該策略會直接在execute方法的調用線程中運行被拒絕的任務；如果執(zhí)行程序已被關閉，則會丟棄該任務
- 設置線程池關閉的時候等待所有任務都完成再繼續(xù)銷毀其他的Bean
- 設置線程池中任務的等待時間，如果超過這個時候還沒有銷毀就強制銷毀，以確保應用最后能夠被關閉，而不是阻塞住
也可以單獨類來配置線程池：
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
/**
* Created by huangguisu on 2020/6/10.
*/
@Configuration
@EnableAsync
public class MyThreadPoolConfig {
private static final int CORE_POOL_SIZE = 10;
private static final int MAX_POOL_SIZE = 20;
private static final int QUEUE_CAPACITY = 200;
public static final String BEAN_EXECUTOR = "bean_executor";
/**
* 事件和情感接口線程池執(zhí)行器配置
* @return 事件和情感接口線程池執(zhí)行器bean
*
*/
@Bean(BEAN_EXECUTOR)
public Executor executor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(CORE_POOL_SIZE);
executor.setMaxPoolSize(MAX_POOL_SIZE);
// 設置隊列容量
executor.setQueueCapacity(QUEUE_CAPACITY);
// 設置線程活躍時間(秒)
executor.setKeepAliveSeconds(60);
executor.setThreadNamePrefix("SE-Pool#Task");
// 設置拒絕策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}
同時注意需要在配置類上添加@EnableAsync，當然也可以在啟動類上添加，表示開啟spring的@@Async
2、異步方法使用線程池
只需要在@Async注解中指定線程池名即可
@Component
public class Task {
//默認使用線程池
@Async
public void doTaskOne() throws Exception {
System.out.println("開始做任務");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任務耗時：" + (end - start) + "毫秒");
}
//根據(jù)Bean Name指定特定線程池
@Async("taskExecutor")
public void doTaskOne() throws Exception {
System.out.println("開始做任務");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任務耗時：" + (end - start) + "毫秒");
}
}
3、通過xml配置定義線程池
Bean文件配置: spring_async.xml
1. 線程的前綴為xmlExecutor
2. 啟動異步線程池配置
<!-- 等價于 @EnableAsync， executor指定線程池 -->
<task:annotation-driven executor="xmlExecutor"/>
<!-- id指定線程池產生線程名稱的前綴 -->
<task:executor
id="xmlExecutor"
pool-size="5-25"
queue-capacity="100"
keep-alive="120"
rejection-policy="CALLER_RUNS"/>
啟動類導入xml文件：
@SpringBootApplication
@ImportResource("classpath:/async/spring_async.xml")
public class AsyncApplicationWithXML {
private static final Logger log = LoggerFactory.getLogger(AsyncApplicationWithXML.class);
public static void main(String[] args) {
log.info("Start AsyncApplication.. ");
SpringApplication.run(AsyncApplicationWithXML.class, args);
}
}
線程池參數(shù)說明
1. ‘id' ： 線程的名稱的前綴
2. ‘pool-size'：線程池的大小。支持范圍”min-max”和固定值（此時線程池core和max sizes相同）
3. ‘queue-capacity' ：排隊隊列長度
4. ‘rejection-policy': 對拒絕的任務處理策略
5. ‘keep-alive' ： 線程?；顣r間（單位秒）
四、異常處理
上面也提到：在調用方法時，可能出現(xiàn)方法中拋出異常的情況。在異步中主要有有兩種異常處理方法：
1. 對于方法返回值是Futrue的異步方法:
a) 、一種是在調用future的get時捕獲異常;
b)、 在異常方法中直接捕獲異常
2. 對于返回值是void的異步方法：通過AsyncUncaughtExceptionHandler處理異常
@Component
public class AsyncException {
/**
* 帶參數(shù)的異步調用 異步方法可以傳入?yún)?shù)
* 對于返回值是void，異常會被AsyncUncaughtExceptionHandler處理掉
* @param s
*/
@Async
public void asyncInvokeWithException(String s) {
log.info("asyncInvokeWithParameter, parementer={}", s);
throw new IllegalArgumentException(s);
}
/**
* 異常調用返回Future
* 對于返回值是Future，不會被AsyncUncaughtExceptionHandler處理，需要我們在方法中捕獲異常并處理
* 或者在調用方在調用Futrue.get時捕獲異常進行處理
*
* @param i
* @return
*/
@Async
public Future<String> asyncInvokeReturnFuture(int i) {
System.out.println("asyncInvokeReturnFuture, parementer={}", i);
Future<String> future;
try {
Thread.sleep(1000 * 1);
future = new AsyncResult<String>("success:" + i);
throw new IllegalArgumentException("a");
} catch (InterruptedException e) {
future = new AsyncResult<String>("error");
} catch(IllegalArgumentException e){
future = new AsyncResult<String>("error-IllegalArgumentException");
}
return future;
}
}
實現(xiàn)AsyncConfigurer接口對異常線程池更加細粒度的控制
a) 創(chuàng)建線程自己的線程池
b) 對void方法拋出的異常處理的類AsyncUncaughtExceptionHandler
@Service
public class MyAsyncConfigurer implements AsyncConfigurer{
@Override
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
threadPool.setCorePoolSize(1);
threadPool.setMaxPoolSize(1);
threadPool.setWaitForTasksToCompleteOnShutdown(true);
threadPool.setAwaitTerminationSeconds(60 * 15);
threadPool.setThreadNamePrefix("MyAsync-");
threadPool.initialize();
return threadPool;
}
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return new MyAsyncExceptionHandler();
}
/**
* 自定義異常處理類
*/
class MyAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
@Override
public void handleUncaughtException(Throwable throwable, Method method, Object... obj) {
System.out.println("Exception message - " + throwable.getMessage());
System.out.println("Method name - " + method.getName());
for (Object param : obj) {
System.out.println("Parameter value - " + param);
}
}
}
}
五、問題
上面也提到：如果不自定義異步方法的線程池默認使用SimpleAsyncTaskExecutor。SimpleAsyncTaskExecutor：不是真的線程池，這個類不重用線程，每次調用都會創(chuàng)建一個新的線程。并發(fā)大的時候會產生嚴重的性能問題。
一般的錯誤OOM：OutOfMemoryError：unable to create new native thread，創(chuàng)建線程數(shù)量太多，占用內存過大.
解決辦法：一般最好使用自定義線程池，做一些特殊策略, 比如自定義拒絕策略，如果隊列滿了，則拒絕處理該任務。
到此這篇關于Spring Boot之@Async異步線程池的文章就介紹到這了,更多相關Spring Boot @Async異步線程池內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
- Spring?Boot異步線程間數(shù)據(jù)傳遞的四種方式
- springboot?正確的在異步線程中使用request的示例代碼
- SpringBoot使用異步線程池實現(xiàn)生產環(huán)境批量數(shù)據(jù)推送
- SpringBoot整合MQTT并實現(xiàn)異步線程調用的問題
- SpringBoot?異步線程間傳遞上下文方式
- SpringBoot獲取HttpServletRequest的3種方式總結
- SpringBoot詳細講解異步任務如何獲取HttpServletRequest
- 在 Spring Boot 中使用異步線程時的 HttpServletRequest 復用問題記錄
相關文章
Spring?Boot教程之提高開發(fā)效率必備工具lombok
這篇文章主要介紹了Spring?Boot教程之提高開發(fā)效率必備工具lombok的相關資料,需要的朋友可以參考下2022-08-08
這篇文章主要介紹了SpringBoot 調度任務及常用任務表達式,需要的朋友可以參考下2017-12-12
java 中Executor, ExecutorService 和 Executors 間的不同
這篇文章主要介紹了java 中Executor, ExecutorService 和 Executors 間的不同的相關資料,需要的朋友可以參考下2017-06-06
深入理解springMVC中的Model和Session屬性
這篇文章主要介紹了深入理解springMVC中的Model和Session屬性，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12
spring squertz是一個強大的定時任務處理方式。下面這篇文章主要介紹了spring利用squertz實現(xiàn)定時任務的相關資料，文中介紹的很詳細，對大家具有一定的參考借鑒價值，需要的朋友們下面來一起看看吧。2017-01-01
這篇文章主要介紹了Java IO流，同時也介紹了流中的一些相關的內容，并且通過大量的案例供大家理解。最后通過一些經(jīng)典的案例幫助大家對前面所學的知識做了一個綜合的應用,需要的朋友可以參考一下2022-05-05
最近在學習搜索方面的東西，需要了解網(wǎng)絡爬蟲方面的知識，雖然有很多開源的強大的爬蟲，但本著學習的態(tài)度，想到之前在做資訊站的時候需要用到爬蟲來獲取一些文章,今天剛好有空就研究了一下.在網(wǎng)上看到了一個demo,使用的是Jsoup,我拿過來修改了一下,有需要的朋友可以參考2016-11-11
Spring?Cloud?Gateway?2.x跨域時出現(xiàn)重復Origin的BUG問題
這篇文章主要介紹了Spring?Cloud?Gateway?2.x跨域時出現(xiàn)重復Origin的BUG問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-04-04
判斷日期經(jīng)常合法出現(xiàn)在IO場景下,下面將盡量使用簡練的思路和代碼呈現(xiàn)給大伙,這篇文章主要給大家介紹了關于Java驗證日期時間字符串是否合法的三種方式,文中通過代碼介紹的非常詳細,需要的朋友可以參考下2023-11-11
SpringBoot解析JSON數(shù)據(jù)的三種方案
JSON(JavaScript Object Notation) 是一種輕量級的數(shù)據(jù)交換格式,易于人閱讀和編寫,同時也易于機器解析和生成,本文給大家介紹了SpringBoot解析JSON數(shù)據(jù)的三種方案,需要的朋友可以參考下2024-03-03