快捷導(dǎo)航

Springboot如何使用@Async實(shí)現(xiàn)異步任務(wù)

更新時(shí)間：2023年09月27日 09:46:34 作者：唐宋xy

這篇文章主要介紹了Springboot如何使用@Async實(shí)現(xiàn)異步任務(wù)問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

前言

在查詢大批量的數(shù)據(jù)的時(shí)候，如果需要查詢多個(gè)表中的數(shù)據(jù)，或者不僅查詢數(shù)據(jù)庫(kù)，還需要取其他的系統(tǒng)中查詢數(shù)據(jù)，然后將所有查詢到的數(shù)據(jù)一起返回，這個(gè)時(shí)候，如果是單線程查詢效率慢，這個(gè)時(shí)候多線程就可以解決這個(gè)查詢效率慢的問(wèn)題，Springboot中提供了@Async注解，一鍵實(shí)現(xiàn)異步操作～

實(shí)戰(zhàn)

一、@Async配置

異步任務(wù)配置類

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
? ? /**
? ? ?* 異步任務(wù)線程池配置
? ? ?* @return
? ? ?*/
? ? @Override
? ? public Executor getAsyncExecutor() {
? ? ? ? ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
? ? ? ? executor.setCorePoolSize(10); // 設(shè)置線程初始化數(shù)量
? ? ? ? executor.setMaxPoolSize(100); // 設(shè)置線程池最大數(shù)量
? ? ? ? executor.setQueueCapacity(100); // 設(shè)置等待隊(duì)列的大小
? ? ? ? executor.setThreadNamePrefix("AsyncExecutorThread-"); // 設(shè)置線程名稱前綴
? ? ? ? executor.initialize(); //如果不初始化，導(dǎo)致找到不到執(zhí)行器
? ? ? ? return executor;
? ? }
? ? /**
? ? ?* 異步任務(wù)異常處理
? ? ?* 如果需要自定義對(duì)異步任務(wù)的異常進(jìn)行處理，則自定義異常處理（實(shí)現(xiàn)AsyncUncaughtExceptionHandler接口），并在這個(gè)方法返回該對(duì)象
? ? ?* @return
? ? ?*/
? ? @Override
? ? public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
? ? ? ? return null;
? ? }
}

二、實(shí)現(xiàn)異步任務(wù)

在需要開(kāi)啟異步的方法上面加上注解@Async表示調(diào)用該方法的時(shí)候，通過(guò)線程池獲取線程去執(zhí)行，如果在Class上加@Async則表示該類的所有方法都是異步執(zhí)行的方法

異步任務(wù)調(diào)用示例

Controller中調(diào)用service中的異步方法更新數(shù)據(jù)，并直接返回

@Slf4j
@Controller
@RequestMapping("/api/dashboard")
public class DashboardController {
? @Autowired
? public DashboardService dashboardService;
? @RequestMapping(value = "/update", method = RequestMethod.PUT)
? @ResponseBody
? public Response update(DashboardLeadsDataVo vo) {
? ? dashboardService.update(vo);
? ? log.info("主線程執(zhí)行完畢，返回結(jié)果")
? ? return new Response();
? }
}

在Service中直接定義異步方法

@Slf4j
@Transactional
@Service
public class DashboardServiceImpl implements DashboardService {
? @Autowired
? private DashboardRepository dashboardRepository;
? /**
? ?* 異步執(zhí)行更新
? ?* @param vo
? ?*/
? @Async
? @Override
? public void update(DashboardLeadsDataVo vo) {
? ?log.info("線程:【{}】執(zhí)行更新【線索數(shù)據(jù)】異步任務(wù)【開(kāi)始】", Thread.currentThread().getName());
? ? DashboardLeadsData dl = new DashboardLeadsData()
? ? BeanUtils.copyProperties(vo, dl);
?? ??? ?dashboardRepository.update(dl)
? ? log.info("線程:【{}】執(zhí)行更新【線索數(shù)據(jù)】異步任務(wù)【結(jié)束】", Thread.currentThread().getName());
? }
}

這樣可以通過(guò)日志看到，開(kāi)啟了新的線程去執(zhí)行@Async注解的方法，主線程在調(diào)用完方法之后就直接可以返回了，子線程在異步執(zhí)行方法。

如果有大量的邏輯運(yùn)算需要進(jìn)行，那么就可以開(kāi)啟多個(gè)子線程去異步執(zhí)行，然后主線程直接返回結(jié)果，響應(yīng)速度快。

三、等待所有子線程完成，主線程返回?cái)?shù)據(jù)

上面的異步任務(wù)沒(méi)有問(wèn)題，但是如果在查詢數(shù)據(jù)的時(shí)候，因?yàn)樾枰樵兌鄠€(gè)數(shù)據(jù)，并組裝之后返回頁(yè)面，所以可以開(kāi)啟多個(gè)異步任務(wù)去執(zhí)行查詢，但是主線程在調(diào)用異步任務(wù)的方法之后，就直接返回了，子線程還沒(méi)有執(zhí)行完成，所以導(dǎo)致返回頁(yè)面的數(shù)據(jù)是沒(méi)有值的。

這個(gè)問(wèn)題需要通過(guò)使用到異步任務(wù)的返回值：Future接口來(lái)實(shí)現(xiàn)返回值和主線程等待子線程執(zhí)行完畢返回

需求：

按照不同的條件查詢多個(gè)表中的數(shù)據(jù)，并返回到頁(yè)面中展示

分析：

因?yàn)橛卸鄠€(gè)表，并且查詢的數(shù)據(jù)量比較大，查詢時(shí)間肯定會(huì)比較長(zhǎng)，那么如果是單線程執(zhí)行，頁(yè)面點(diǎn)擊查詢的時(shí)候，頁(yè)面白屏或者Loading的時(shí)間會(huì)比較長(zhǎng)，所以需要開(kāi)啟多線程異步去查詢，并在多個(gè)線程都查詢完數(shù)據(jù)之后主線程統(tǒng)一返回頁(yè)面渲染

實(shí)現(xiàn)：

Controller中直接調(diào)用service中的方法
@Controller
@RequestMapping("/api/dashboard")
public class DashboardController {
? ? @Autowired
? ? public DashboardService dashboardService;
? ? /**
? ? ?* 根據(jù)時(shí)間查詢顯示的各種線索量
? ? ?* @return
? ? ?*/
? ? @RequestMapping(value = "/cardsData", method = RequestMethod.GET)
? ? @ResponseBody
? ? public Response<DashboardDataDto> getCardsData() {
? ? ? ? DashboardDataDto dto = dashboardService.getCardsDataByDate();
? ? ? ? return new Response<>(dto);
? ? }
}

由service中調(diào)用多個(gè)異步任務(wù)方法查詢數(shù)據(jù)并統(tǒng)一返回Controller

@Slf4j
@Transactional
@Service
public class DashboardServiceImpl implements DashboardService {
? ? @Autowired
? ? private DashboardRepository dashboardRepository;
?/**
? ? ?* 查詢各種線索量
? ? ?*/
? ? @Override
? ? public DashboardDataDto getCardsDataByDate() {
? ? ? ? List<Future> futureList = new ArrayList<>();
? ? ? ? DashboardDataDto dto = new DashboardDataDto();
? ? ? ? // 從Spring容器中獲取當(dāng)前類的對(duì)象，不能直接調(diào)用自身方法，否則@Async注解無(wú)效 或者在其他spring管理類中調(diào)用
? ? ? ? DashboardServiceImpl springProxyBean = SpringUtil.getBean(DashboardServiceImpl.class);
? ? ? ? // 獲取總數(shù)據(jù)
? ? ? ? Future<String> totalFuture = springProxyBean.getTotalCount(dto);
? ? ? ? futureList.add(totalFuture);
? ? ? ? // 獲取年份數(shù)據(jù)
? ? ? ? Future<String> yearFuture = springProxyBean.getTerritoryYearCount(dto);
? ? ? ? futureList.add(yearFuture);
? ? ? ?? ?// 判斷異步任務(wù)中的所有子線程是否執(zhí)行完成
? ? ? ? // Future接口不僅可以用來(lái)判斷線程的狀態(tài)，還可以獲取到異步任務(wù)執(zhí)行的返回結(jié)果
? ? ? ? while (true) {
? ? ? ? ? ? if(futureList.size() > 0) {
? ? ? ? ? ? ? ? boolean isAllThreadDone = true;
? ? ? ? ? ? ? ? for (Future future : futureList) {
? ? ? ? ? ? ? ? ? ? // 判斷是否所有的線程都已經(jīng)執(zhí)行完成
? ? ? ? ? ? ? ? ? ? if(future == null || !future.isDone()) {
? ? ? ? ? ? ? ? ? ? ? ? isAllThreadDone = false;
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ?/** 如果需要獲取異步任務(wù)的返回值，則先判斷是否執(zhí)行完成
? ? ? ? ? ? ? ? ? ? ?* if (future.isDone()) {
? ? ? ? ? ? ? ? ? ? ?* ? ?future.get();
? ? ? ? ? ? ? ? ? ? ?* ?}
? ? ? ? ? ? ? ? ? ? ?*/
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? // 都已經(jīng)執(zhí)行完成則break
? ? ? ? ? ? ? ? if(isAllThreadDone) {
? ? ? ? ? ? ? ? ? ? break;
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? log.info("主線程結(jié)束。。。。。。。");
? ? ? ? return dto;
? ? }?? ?
? ?? ?/**
? ? ?* 查詢總數(shù)據(jù)
? ? ?* @param dto
? ? ?*/
? ? @Async
? ? public Future<String> getTotalCount(DashboardDataDto dto) {
? ? ? ? log.info("線程:【{}】執(zhí)行查詢【總數(shù)據(jù)】異步任務(wù)【開(kāi)始】", Thread.currentThread().getName());
? ? ? ? // 查詢領(lǐng)界累計(jì)線索量
? ? ? ? List<Object> totalLeadsCount = dashboardRepository.findTotalLeadsCount();
? ? ? ? if(!CollectionUtils.isEmpty(totalLeadsCount)) {
? ? ? ? ? ? dto.setTotalLeadsCount(((Number) totalLeadsCount.get(0)).intValue());
? ? ? ? }
? ? ? ? log.info("線程:【{}】執(zhí)行查詢【總數(shù)據(jù)】異步任務(wù)【結(jié)束】", Thread.currentThread().getName());
? ? ? ?? ?// 只可以返回Future接口實(shí)現(xiàn)類，并且將需要返回的數(shù)據(jù)傳遞
? ? ? ? return new AsyncResult<>(Thread.currentThread().getName());
? ? }
? ? /**
? ? ?* 查詢年份領(lǐng)界線索數(shù)據(jù)量
? ? ?* @param dto
? ? ?*/
? ? @Async
? ? public Future<String> getYearCount(DashboardDataDto dto) {
? ? ? ? log.info("線程:【{}】執(zhí)行查詢【年份數(shù)據(jù)量】異步任務(wù)【開(kāi)始】", Thread.currentThread().getName());
? ? ? ? String currentYearStart = DateUtils.yearStart(0);
? ? ? ? String nextYearStart = DateUtils.yearStart(1);
? ? ? ? // 查詢年份累計(jì)線索量
? ? ? ? List<Object> yearsCountList = dashboardRepository.findYearsCountByDate(currentYearStart, nextYearStart);
? ? ? ? if(!CollectionUtils.isEmpty(yearsCountList)) {
? ? ? ? ? ? dto.setYearsCount(((Number) yearsCountList.get(0)).intValue());
? ? ? ? }
? ? ? ? log.info("線程:【{}】執(zhí)行查詢【年份數(shù)據(jù)量】異步任務(wù)【結(jié)束】", Thread.currentThread().getName());
? ? ? ? return new AsyncResult<>(Thread.currentThread().getName());
? ? }
}

Service中用到的 SpringUtil

如果是自身調(diào)用自身的異步方法（如果是調(diào)用其他的spring管理類中異步方法，無(wú)須通過(guò)手動(dòng)獲取該對(duì)象，直接調(diào)用即可），需要通過(guò)Spring的容器中獲取到當(dāng)前類的對(duì)象，否則自身調(diào)用方法無(wú)效，因?yàn)闆](méi)有走spring的代理，@Async注解無(wú)法生效

@Component
public class SpringUtil implements ApplicationContextAware {
? ? private static ApplicationContext applicationContext = null;
? ? public static ApplicationContext getApplicationContext() {
? ? ? ? return applicationContext;
? ? }
? ? /**
? ? ?* Spring容器啟動(dòng)后，會(huì)把 applicationContext 給自動(dòng)注入進(jìn)來(lái)，然后我們把 applicationContext
? ? ?* ?賦值到靜態(tài)變量中，方便后續(xù)拿到容器對(duì)象
? ? ?* @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext)
? ? ?*/
? ? @Override
? ? public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
? ? ? ? SpringUtil.applicationContext = applicationContext;
? ? }
? ? @SuppressWarnings("unchecked")
? ? public static <T> T getBean(String beanId) {
? ? ? ? return (T) applicationContext.getBean(beanId);
? ? }
? ? public static <T> T getBean(Class<T> requiredType) {
? ? ? ? return (T) applicationContext.getBean(requiredType);
? ? }
}

調(diào)用說(shuō)明

經(jīng)測(cè)試，主線程的log，會(huì)在所有的需要等待的子線程執(zhí)行完畢之后，才會(huì)返回到Controller，返回到頁(yè)面中的數(shù)據(jù)就是所有線程已查詢出來(lái)的數(shù)據(jù)對(duì)象，響應(yīng)速度加快

TODO

實(shí)現(xiàn)多個(gè)線程同時(shí)開(kāi)始執(zhí)行某個(gè)任務(wù)或者多個(gè)線程都同時(shí)處于等待狀態(tài)時(shí)開(kāi)始執(zhí)行某個(gè)任務(wù)，可以使用CountDownLatch、CyclicBarrier來(lái)實(shí)現(xiàn)，可以用來(lái)實(shí)現(xiàn)等待所有的子線程返回?cái)?shù)據(jù)，主線程再返回功能，后續(xù)再說(shuō)明～～

四、@ASync無(wú)效說(shuō)明

沒(méi)有在@SpringBootApplication啟動(dòng)類或者異步任務(wù)配置類當(dāng)中添加注解@EnableAsync注解。

異步方法使用注解@Async的返回值只能為void或者Future。

沒(méi)有走Spring的代理類。因?yàn)锧Transactional和@Async注解的實(shí)現(xiàn)都是基于Spring的AOP，而AOP的實(shí)現(xiàn)是基于動(dòng)態(tài)代理模式實(shí)現(xiàn)的。那么注解失效的原因就很明顯了，有可能因?yàn)檎{(diào)用方法的是對(duì)象本身而不是代理對(duì)象，因?yàn)闆](méi)有經(jīng)過(guò)Spring容器。

注解的方法必須是public方法，編譯時(shí)非public方法會(huì)報(bào)錯(cuò)

如果需要從類的內(nèi)部調(diào)用，需要先獲取其代理類