1. 介绍

@Async 注解在 service 内的方法上(@EnableAsync开启异步)，可以实现在 controller 的异步调用，调用的被 @Async 注解的方法会在一个单独线程内运行，适合即使返回，异步解耦，service 慢慢去处理

@Async 注解的方法只能 返回 void 或者 future 类型的返回值，其他值会使 注解无效，因为不能异步执行

被 @Async 的方法在独立线程调用，不能被 @ControllerAdvice 全局异常处理器捕获，所以需要自己设置异常处理

  1. import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
  2. import org.springframework.context.annotation.Configuration;
  3. import org.springframework.scheduling.annotation.AsyncConfigurer;
  4. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  6. import java.util.concurrent.Executor;
  8. /**
  9.  * @description:
  10.  * @author: qinyu
  11.  * @Date: 2019/8/22 21:09
  12.  */
  13. @Configuration
  14. public class ThreadAsyncConfiguration implements AsyncConfigurer {
  16.     @Override
  17.     public Executor getAsyncExecutor() {
  18.         ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
  19.         //设置核心线程数
  20.         threadPool.setCorePoolSize(10);
  21.         //设置最大线程数
  22.         threadPool.setMaxPoolSize(100);
  23.         //线程池所使用的缓冲队列
  24.         threadPool.setQueueCapacity(10);
  25.         //等待任务在关机时完成--表明等待所有线程执行完
  26.         threadPool.setWaitForTasksToCompleteOnShutdown(true);
  27.         // 等待时间 （默认为0，此时立即停止），并没等待xx秒后强制停止
  28.         threadPool.setAwaitTerminationSeconds(60);
  29.         //  线程名称前缀s
  30.         threadPool.setThreadNamePrefix("MyAsync-");
  31.         // 初始化线程
  32.         threadPool.initialize();
  33.         return threadPool;
  34.     }
  36.     @Override
  37.     public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
  38.         return new SpringAsyncExceptionHandler();
  39.     }
  40. }

SpringAsyncExceptionHandler 类

  1. @Slf4j
  2. public class SpringAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
  3.     @Override
  4.     public void handleUncaughtException(Throwable throwable, Method method, Object... objects) {
  5.         log.error("Exception occurs in async method", throwable.getMessage());
  6.     }
  7. }

2. 使用实例一

springboot 使用异步注解 @Async

  1. @Component
  2. public class AsyncExceptionDemo {
  3.     private static final Logger log = LoggerFactory.getLogger(AsyncExceptionDemo.class);
  5.     /**
  6.      * 最简单的异步调用，返回值为void
  7.      */
  8.     @Async
  9.     public void asyncInvokeSimplest() {
  10.         log.info("asyncSimplest");
  11.     }
  13.     /**
  14.      * 带参数的异步调用 异步方法可以传入参数
  15.      *  对于返回值是void，异常会被AsyncUncaughtExceptionHandler处理掉
  16.      * @param s
  17.      */
  18.     @Async
  19.     public void asyncInvokeWithException(String s) {
  20.         log.info("asyncInvokeWithParameter, parementer={}", s);
  21.         throw new IllegalArgumentException(s);
  22.     }
  24.     /**
  25.      * 异常调用返回Future
  26.      *  对于返回值是Future，不会被AsyncUncaughtExceptionHandler处理，需要我们在方法中捕获异常并处理
  27.      *  或者在调用方在调用Futrue.get时捕获异常进行处理
  28.      * 
  29.      * @param i
  30.      * @return
  31.      */
  32.     @Async
  33.     public Future<String> asyncInvokeReturnFuture(int i) {
  34.         log.info("asyncInvokeReturnFuture, parementer={}", i);
  35.         Future<String> future;
  36.         try {
  37.             Thread.sleep(1000 * 1);
  38.             future = new AsyncResult<String>("success:" + i);
  39.             throw new IllegalArgumentException("a");
  40.         } catch (InterruptedException e) {
  41.             future = new AsyncResult<String>("error");
  42.         } catch(IllegalArgumentException e){
  43.             future = new AsyncResult<String>("error-IllegalArgumentException");
  44.         }
  45.         return future;
  46.     }
  48. }

启动异步调用

  1. @SpringBootApplication
  2. @EnableAsync // 启动异步调用
  3. public class AsyncApplicationWithAsyncConfigurer {
  4.     private static final Logger log = LoggerFactory.getLogger(AsyncApplicationWithAsyncConfigurer.class);
  6.     public static void main(String[] args) {
  7.         log.info("Start AsyncApplication.. ");
  8.         SpringApplication.run(AsyncApplicationWithAsyncConfigurer.class, args);
  9.     }
  10. }

3. 结合 CountDownLatch 异步获取结果

调用

  1. @Override
  2. public Result queryQuestionUseCountData(List<String> queIdList) throws ServiceException {
  3.     try {
  4.         if (ListUtils.isEmpty(queIdList)) {
  5.             throw new Exception("试题id不能为空");
  6.         }
  7.         CountDownLatch countDownLatch = new CountDownLatch(3);
  8.         //查询相似题数量
  9.         Future<Map> relativeQuestionCountMapFuture = questionAsyncService.queryRelativeQuestionCount(queIdList, countDownLatch);
  10.         //查询试题使用数据
  11.         Future<List> questionStateCountListFuture = questionAsyncService.queryQuestionStateCountList(queIdList, countDownLatch);
  12.         // 查询网校使用数量
  13.         Future<List> onlineSchoolPaperNumberFuture = questionAsyncService.queryOnlineSchoolUseNumList(queIdList, countDownLatch);
  15.         try {
  16.             countDownLatch.await(10, TimeUnit.SECONDS);
  17.         } catch (InterruptedException e) {
  18.             throw new ServiceException("子线程等待异常");
  19.         }
  20.         // 封装试题使用数据返回结果
  21.         Map<String, QueUseCountVo> returnMap = packageQuestionUseCountDataResult(queIdList, relativeQuestionCountMapFuture.get(), questionStateCountListFuture.get(), onlineSchoolPaperNumberFuture.get());
  22.         // 根据配置字参数对使用/作答/正确率三个参数的显示进行过滤
  23.         filterQuestionCountData(returnMap);
  24.         return Result.buildSuccessResult(returnMap);
  25.     } catch (Exception e) {
  26.         log.error("查询试题使用数据失败.", e);
  27.         throw new ServiceException(e);
  28.     }
  29. }

结合 @Async 异步方法

  1. /**
  2.  * @description: @Async 标注的方法和调用者不能在同一个类中，会产生代理绕过问题，因此创建此类放置异步方法
  3.  * @author: zcq
  4.  * @date: 2020/8/4 4:43 下午
  5.  */
  6. @Service
  7. @Slf4j
  8. public class QuestionServiceAsyncImpl implements IQuestionAsyncService {
  10.     @Autowired
  11.     private DataCenterRestClient dataCenterRestClient;
  12.     @Autowired
  13.     private EsRestClient esRestClient;
  14.     @Autowired
  15.     private EpAnalyRestClient epAnalyRestClient;
  17.     /**
  18.      * @description: 查询相似题数量
  19.      * @author: zcq
  20.      * @date: 2020/8/3 11:16 上午
  21.      */
  22.     @Override
  23.     @Async
  24.     public Future<Map> queryRelativeQuestionCount(List<String> queIdList, CountDownLatch countDownLatch) throws Exception {
  25.         Map<String, Object> relativeQuestionCountMap = new HashMap<>();
  26.         try {
  27.             Result relativeQuestionCountResult = dataCenterRestClient.queryRelativeQuestionCount(queIdList);
  28.             if (relativeQuestionCountResult.getSuccess()) {
  29.                 relativeQuestionCountMap = (Map<String, Object>) relativeQuestionCountResult.getData();
  30.             }
  31.             return new AsyncResult(relativeQuestionCountMap);
  32.         } catch (RestClientException e) {
  33.             log.error("通过DataCenterRestClient 查询相似题数量失败.", e);
  34.             throw new Exception(e);
  35.         } finally {
  36.             countDownLatch.countDown();
  37.         }
  38.     }
  40.     /**
  41.      * @description: 查询试题使用数据
  42.      * @author: zcq
  43.      * @date: 2020/8/3 11:16 上午
  44.      */
  45.     @Override
  46.     @Async
  47.     public Future<List> queryQuestionStateCountList(List<String> queIdList, CountDownLatch countDownLatch) throws Exception {
  48.         List<QueUseCountVo> questionStateCountList = new ArrayList<>();
  49.         try {
  50.             Result questionStateCountResult = esRestClient.queryQuestionState(queIdList);
  51.             if (questionStateCountResult.getSuccess() && questionStateCountResult.getData() != null) {
  52.                 questionStateCountList = JSON.parseArray(JSON.toJSONString(questionStateCountResult.getData()), QueUseCountVo.class);
  53.             }
  54.             return new AsyncResult(questionStateCountList);
  55.         } catch (RestClientException e) {
  56.             log.error("通过EsRestClient 获取试题统计信息失败.", e);
  57.             throw new Exception(e);
  58.         } finally {
  59.             countDownLatch.countDown();
  60.         }
  61.     }
  63.     /**
  64.      * @description: 查询试题网校的使用数据
  65.      * @author: zcq
  66.      * @date: 2020/8/3 11:16 上午
  67.      */
  68.     @Override
  69.     @Async
  70.     public Future<List> queryOnlineSchoolUseNumList(List<String> queIdList, CountDownLatch countDownLatch) throws Exception {
  71.         List<QueUseCountVo> onlineSchoolUseNumList = new ArrayList<>();
  72.         try {
  73.             Result questionWxUseCountResult = epAnalyRestClient.queryOnlineSchoolUseNumList(queIdList);
  74.             if (questionWxUseCountResult.getSuccess() && questionWxUseCountResult.getData() != null) {
  75.                 onlineSchoolUseNumList = JSON.parseArray(JSON.toJSONString(questionWxUseCountResult.getData()), QueUseCountVo.class);
  76.             }
  77.             return new AsyncResult(onlineSchoolUseNumList);
  78.         } catch (RestClientException e) {
  79.             log.error("调用接口查询网校试题使用数据失败.", e);
  80.             throw new Exception(e);
  81.         } finally {
  82.             countDownLatch.countDown();
  83.         }
  84.     }
  85. }

注意：

@Async 异步方法不要和同步调用方法写在同一个类中，即异步方法和调用者不要再同一个类中，会产生代理绕过问题。