springboot @Async 註解如何實現方法異步
@Async註解如何實現方法異步
處理大批量數據的時候,效率很慢。所以考慮一下使用多線程。
剛開始自己手寫的一套,用瞭線程池啟動固定的線程數進行跑批。但是後來老大考慮到自己手寫的風險不好控制,所以使用spring的方法。
這裡沒有詳細介紹,隻有簡單的demo,隻會用,不懂原理:
一、springboot的App類需要的註解
package com.xxx.xxx.xxx; import java.util.concurrent.ThreadPoolExecutor; import org.springframework.boot.web.support.SpringBootServletInitializer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.task.TaskExecutor; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; /** * 類功能說明:服務生產者啟動類 * <p> * <strong></strong> * </p> * * @version * @author * @since 1.8 */ @Configuration @EnableAsync public class Application extends SpringBootServletInitializer { @Bean public TaskExecutor taskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); // 設置核心線程數 executor.setCorePoolSize(5); // 設置最大線程數 executor.setMaxPoolSize(60); // 設置隊列容量 executor.setQueueCapacity(20); // 設置線程活躍時間(秒) executor.setKeepAliveSeconds(60); // 設置默認線程名稱 executor.setThreadNamePrefix("what-"); // 設置拒絕策略 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 等待所有任務結束後再關閉線程池 executor.setWaitForTasksToCompleteOnShutdown(true); return executor; } }
springboot的App類,很簡單,就能使用很多東西。
二、service層的註解
package com.xxx.xxx.service.impl; import java.util.concurrent.Future; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.AsyncResult; import org.springframework.stereotype.Service; import com.xxx.xxx.service.XXXAsyncService ; @Service public class XXXAsyncServiceImpl implements XXXAsyncService { @Async public Future<Long> rtn1() throws Exception { //do something //有返回值的時候,可以返回string,long之類的。 return new AsyncResult<>(1); } @Async public void rtn2() throws Exception { //do something //這個可以沒有返回值. } }
三、調用層
package com.xxx.xxx.controller; import java.util.concurrent.Future; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.RestController; import com.xxx.xxx.service.XXXAsyncService; @RestController @RequestMapping(value="/xxx") public class XXXAsyncController { @Autowired private XXXAsyncService xxxAsyncService; /** * 這裡調用異步方法 */ @RequestMapping(value = "/xxx") public void dodo() throws Exception { int threads = 10;//十個線程 List<Future<Long>> list = new ArrayList<>(); for(int i = 0;i < threads; i++){ //這裡循環調用異步方法。 //如果存在大量數據,可以在這裡把數據切片,然後循環調用,分批處理數據。效率杠杠的。 list .add(xxxAsyncService.rtn1()); } long count = 0; for(Future<Long> l : tsfCountList) { //異步調用需要返回值的時候,這裡可以把返回值都放入到list集合中,然後可以統一處理。 這裡的get()是阻塞的,因為需要所以異步方法返回,在繼續執行。 count += l.get(); } System.out.println("調用次數:" + count); } }
這些代碼全是手寫,記錄下來,以後用的時候,省的忘瞭,查起來麻煩。。
異步註解@Async的使用以及註意事項
第一步開啟異步
@Configuration @EnableAsync public class SpringAsyncConfig { ... }
默認情況下,@EnableAsync檢測Spring的@Async註釋和EJB 3.1 javax. EJB .異步;此選項還可用於檢測其他用戶定義的註釋類型。(也可以在SpringBoot的啟動類上直接加@EnableAsync註解)
在 Spring 中,用 @Async 註解指定的方法,該方法被調用時會以異步的方式執行。而如果沒有在 @Async 註解中指定線程池,就會使用默認的線程池。默認的線程池為 SimpleAsyncTaskExecutor 。
該線程池不會復用線程,每有一個新任務被提交,該線程池就會創建一個新的線程實例用於執行任務。下面為相關的代碼:
protected void doExecute(Runnable task) { Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task)); thread.start(); }
而如果想要指定線程池,可以通過在 @Async 註解中的 value 參數中指定所要使用的線程池的 Bean Name 。另一種方法是是一個實現瞭 AsyncConfigurer 接口或是繼承其默認適配器類 AsyncConfigurerSupport 的配置類,這樣 @Async 註解的方法就會使用指定的自定義的線程池。
使用@Async註解的話采用的是springBoot默認的線程池,不過一般我們會自定義線程池(因為比較靈活),配置方式有:
- 使用 xml 文件配置的方式
- 使用Java代碼結合@Configuration進行配置(推薦使用)
下面顯示配置線程的代碼實現
package com.deppon.ptos.load.config; import lombok.extern.slf4j.Slf4j; 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; /** * @Description: 異步線程管理 * @Author: LYH * @CreateDate: 2019/6/27 8:54 * @Version: 1.0 * @JDK: 1.8 */ @Configuration @EnableAsync @Slf4j public class ExecutorConfig { @Value("${async.executor.thread.core_pool_size}") private int corePoolSize; @Value("${async.executor.thread.max_pool_size}") private int maxPoolSize; @Value("${async.executor.thread.queue_capacity}") private int queueCapacity; @Value("${async.executor.thread.name.prefix}") private String namePrefix; @Bean(name = "asyncServiceExecutor") public Executor asyncServiceExecutor() { log.info("start asyncServiceExecutor"); ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor(); //配置核心線程數 executor.setCorePoolSize(corePoolSize); //配置最大線程數 executor.setMaxPoolSize(maxPoolSize); //配置隊列大小 executor.setQueueCapacity(queueCapacity); //配置線程池中的線程的名稱前綴 executor.setThreadNamePrefix(namePrefix); // rejection-policy:當pool已經達到max size的時候,如何處理新任務 // CALLER_RUNS:不在新線程中執行任務,而是有調用者所在的線程來執行 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); //執行初始化 executor.initialize(); return executor; } }
package com.deppon.ptos.load.config; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import org.springframework.util.concurrent.ListenableFuture; import java.util.concurrent.Callable; import java.util.concurrent.Future; import java.util.concurrent.ThreadPoolExecutor; /** * @Description: 打印異步線程的執行情況 使用Callbale Future 來返回線程的信息 * @Author: 633805 LYH * @CreateDate: 2019/6/27 8:59 * @Version: 1.0 * @JDK: 1.8 */ @Component @Slf4j public class VisiableThreadPoolTaskExecutor extends ThreadPoolTaskExecutor { private void showThreadPoolInfo(String prefix) { ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor(); if (null == threadPoolExecutor) { return; } log.info("{}, {},taskCount [{}], completedTaskCount [{}], activeCount [{}], queueSize [{}]", this.getThreadNamePrefix(), prefix, threadPoolExecutor.getTaskCount(), threadPoolExecutor.getCompletedTaskCount(), threadPoolExecutor.getActiveCount(), threadPoolExecutor.getQueue().size()); } @Override public void execute(Runnable task) { showThreadPoolInfo("1. do execute"); super.execute(task); } @Override public void execute(Runnable task, long startTimeout) { showThreadPoolInfo("2. do execute"); super.execute(task, startTimeout); } @Override public Future<?> submit(Runnable task) { showThreadPoolInfo("1. do submit"); return super.submit(task); } @Override public <T> Future<T> submit(Callable<T> task) { showThreadPoolInfo("2. do submit"); return super.submit(task); } @Override public ListenableFuture<?> submitListenable(Runnable task) { showThreadPoolInfo("1. do submitListenable"); return super.submitListenable(task); } @Override public <T> ListenableFuture<T> submitListenable(Callable<T> task) { showThreadPoolInfo("2. do submitListenable"); return super.submitListenable(task); } }
使用:
@Async("asyncServiceExecutor")
到這一步,異步就算開啟瞭。
下面主要說一說錯誤的
使用@Async導致異步不成功的情況
如下方式會使@Async失效
- 異步方法使用static修飾
- 異步類沒有使用@Component註解(或其他註解)導致spring無法掃描到異步類
- 異步方法不能與被調用的異步方法在同一個類中
- 類中需要使用@Autowired或@Resource等註解自動註入,不能自己手動new對象
- 如果使用SpringBoot框架必須在啟動類中增加@EnableAsync註解
以上為個人經驗,希望能給大傢一個參考,也希望大傢多多支持WalkonNet。
推薦閱讀:
- Spring Boot使用線程池處理上萬條數據插入功能
- Springboot 如何使用@Async整合線程池
- 基於ThreadPoolTaskExecutor的使用說明
- 關於java中@Async異步調用詳細解析附代碼
- SpringBoot定時任務多線程實現示例