Spring系列第37篇：@EnableAsync&@Async实现方法异步调用

star2017 1年前 ⋅ 501 阅读

1、本文内容

详解 @EnableAsync & @Async，主要分下面几个点进行介绍。

  1. 作用
  2. 用法
  3. 获取异步执行结果
  4. 自定义异步执行的线程池
  5. 自定义异常处理
  6. 线程隔离
  7. 源码 & 原理

2、作用

spring容器中实现bean方法的异步调用。

比如有个logService的bean，logservice中有个log方法用来记录日志，当调用logService.log(msg)的时候，希望异步执行，那么可以通过@EnableAsync & @Async来实现。

3、用法

2步

  1. 需要异步执行的方法上面使用@Async注解标注，若bean中所有的方法都需要异步执行，可以直接将@Async加载类上。
  2. 将@EnableAsync添加在spring配置类上，此时@Async注解才会起效。

常见2种用法

  1. 无返回值的
  2. 可以获取返回值的

4、无返回值的

用法

方法返回值不是Future类型的，被执行时，会立即返回，并且无法获取方法返回值，如：

  1. @Async
  2. public void log(String msg) throws InterruptedException {
  3.     System.out.println("开始记录日志," + System.currentTimeMillis());
  4.     //模拟耗时2秒
  5.     TimeUnit.SECONDS.sleep(2);
  6.     System.out.println("日志记录完毕," + System.currentTimeMillis());
  7. }

案例

实现日志异步记录的功能。

LogService.log方法用来异步记录日志，需要使用@Async标注

  1. package com.javacode2018.async.demo1;
  3. import org.springframework.scheduling.annotation.Async;
  4. import org.springframework.stereotype.Component;
  6. import java.util.concurrent.TimeUnit;
  8. @Component
  9. public class LogService {
  10.     @Async
  11.     public void log(String msg) throws InterruptedException {
  12.         System.out.println(Thread.currentThread() + "开始记录日志," + System.currentTimeMillis());
  13.         //模拟耗时2秒
  14.         TimeUnit.SECONDS.sleep(2);
  15.         System.out.println(Thread.currentThread() + "日志记录完毕," + System.currentTimeMillis());
  16.     }
  17. }

来个spring配置类，需要加上@EnableAsync开启bean方法的异步调用.

  1. package com.javacode2018.async.demo1;
  3. import org.springframework.context.annotation.ComponentScan;
  4. import org.springframework.context.annotation.EnableAspectJAutoProxy;
  5. import org.springframework.scheduling.annotation.EnableAsync;
  7. @ComponentScan
  8. @EnableAsync
  9. public class MainConfig1 {
  10. }

测试代码

  1. package com.javacode2018.async;
  3. import com.javacode2018.async.demo1.LogService;
  4. import com.javacode2018.async.demo1.MainConfig1;
  5. import org.junit.Test;
  6. import org.springframework.context.annotation.AnnotationConfigApplicationContext;
  8. import java.util.concurrent.TimeUnit;
  10. public class AsyncTest {
  12.     @Test
  13.     public void test1() throws InterruptedException {
  14.         AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
  15.         context.register(MainConfig1.class);
  16.         context.refresh();
  17.         LogService logService = context.getBean(LogService.class);
  18.         System.out.println(Thread.currentThread() + " logService.log start," + System.currentTimeMillis());
  19.         logService.log("异步执行方法!");
  20.         System.out.println(Thread.currentThread() + " logService.log end," + System.currentTimeMillis());
  22.         //休眠一下，防止@Test退出
  23.         TimeUnit.SECONDS.sleep(3);
  24.     }
  26. }

运行输出

  1. Thread[main,5,main] logService.log start,1595223990417
  2. Thread[main,5,main] logService.log end,1595223990432
  3. Thread[SimpleAsyncTaskExecutor-1,5,main]开始记录日志,1595223990443
  4. Thread[SimpleAsyncTaskExecutor-1,5,main]日志记录完毕,1595223992443

前2行输出，可以看出logService.log立即就返回了，后面2行来自于log方法，相差2秒左右。

前面2行在主线程中执行，后面2行在异步线程中执行。

5、获取异步返回值

用法

若需取异步执行结果，方法返回值必须为Future类型，使用spring提供的静态方法org.springframework.scheduling.annotation.AsyncResult#forValue创建返回值，如：

  1. public Future<String> getGoodsInfo(long goodsId) throws InterruptedException {
  2.     return AsyncResult.forValue(String.format("商品%s基本信息!", goodsId));
  3. }

案例

场景：电商中商品详情页通常会有很多信息：商品基本信息、商品描述信息、商品评论信息，通过3个方法来或者这几个信息。

这3个方法之间无关联，所以可以采用异步的方式并行获取，提升效率。

下面是商品服务，内部3个方法都需要异步，所以直接在类上使用@Async标注了，每个方法内部休眠500毫秒，模拟一下耗时操作。

  1. package com.javacode2018.async.demo2;
  3. import org.springframework.scheduling.annotation.Async;
  4. import org.springframework.scheduling.annotation.AsyncResult;
  5. import org.springframework.stereotype.Component;
  7. import java.util.Arrays;
  8. import java.util.List;
  9. import java.util.concurrent.Future;
  10. import java.util.concurrent.TimeUnit;
  12. @Async
  13. @Component
  14. public class GoodsService {
  15.     //模拟获取商品基本信息，内部耗时500毫秒
  16.     public Future<String> getGoodsInfo(long goodsId) throws InterruptedException {
  17.         TimeUnit.MILLISECONDS.sleep(500);
  18.         return AsyncResult.forValue(String.format("商品%s基本信息!", goodsId));
  19.     }
  21.     //模拟获取商品描述信息，内部耗时500毫秒
  22.     public Future<String> getGoodsDesc(long goodsId) throws InterruptedException {
  23.         TimeUnit.MILLISECONDS.sleep(500);
  24.         return AsyncResult.forValue(String.format("商品%s描述信息!", goodsId));
  25.     }
  27.     //模拟获取商品评论信息列表，内部耗时500毫秒
  28.     public Future<List<String>> getGoodsComments(long goodsId) throws InterruptedException {
  29.         TimeUnit.MILLISECONDS.sleep(500);
  30.         List<String> comments = Arrays.asList("评论1", "评论2");
  31.         return AsyncResult.forValue(comments);
  32.     }
  33. }

来个spring配置类，需要加上@EnableAsync开启bean方法的异步调用.

  1. package com.javacode2018.async.demo2;
  3. import org.springframework.context.annotation.ComponentScan;
  4. import org.springframework.scheduling.annotation.EnableAsync;
  6. @ComponentScan
  7. @EnableAsync
  8. public class MainConfig2 {
  9. }

测试代码

  1. @Test
  2. public void test2() throws InterruptedException, ExecutionException {
  3.     AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
  4.     context.register(MainConfig2.class);
  5.     context.refresh();
  6.     GoodsService goodsService = context.getBean(GoodsService.class);
  8.     long starTime = System.currentTimeMillis();
  9.     System.out.println("开始获取商品的各种信息");
  11.     long goodsId = 1L;
  12.     Future<String> goodsInfoFuture = goodsService.getGoodsInfo(goodsId);
  13.     Future<String> goodsDescFuture = goodsService.getGoodsDesc(goodsId);
  14.     Future<List<String>> goodsCommentsFuture = goodsService.getGoodsComments(goodsId);
  16.     System.out.println(goodsInfoFuture.get());
  17.     System.out.println(goodsDescFuture.get());
  18.     System.out.println(goodsCommentsFuture.get());
  20.     System.out.println("商品信息获取完毕,总耗时(ms)：" + (System.currentTimeMillis() - starTime));
  22.     //休眠一下，防止@Test退出
  23.     TimeUnit.SECONDS.sleep(3);
  24. }

运行输出

  1. 开始获取商品的各种信息
  2. 商品1基本信息!
  3. 商品1描述信息!
  4. [评论1, 评论2]
  5. 商品信息获取完毕,总耗时(ms)：525

3个方法总计耗时500毫秒左右。

如果不采用异步的方式，3个方法会同步执行，耗时差不多1.5秒，来试试，将GoodsService上的@Async去掉，然后再次执行测试案例，输出

  1. 开始获取商品的各种信息
  2. 商品1基本信息!
  3. 商品1描述信息!
  4. [评论1, 评论2]
  5. 商品信息获取完毕,总耗时(ms)：1503

这个案例大家可以借鉴一下，按照这个思路可以去优化一下你们的代码，方法之间无关联的可以采用异步的方式，并行去获取，最终耗时为最长的那个方法，整体相对于同步的方式性能提升不少。

6、自定义异步执行的线程池

默认情况下，@EnableAsync使用内置的线程池来异步调用方法，不过我们也可以自定义异步执行任务的线程池。

有2种方式来自定义异步处理的线程池

方式1

在spring容器中定义一个线程池类型的bean，bean名称必须是taskExecutor

  1. @Bean
  2. public Executor taskExecutor() {
  3.     ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  4.     executor.setCorePoolSize(10);
  5.     executor.setMaxPoolSize(100);
  6.     executor.setThreadNamePrefix("my-thread-");
  7.     return executor;
  8. }

方式2

定义一个bean，实现AsyncConfigurer接口中的getAsyncExecutor方法，这个方法需要返回自定义的线程池，案例代码：

  1. package com.javacode2018.async.demo3;
  3. import com.javacode2018.async.demo1.LogService;
  4. import org.springframework.beans.factory.annotation.Qualifier;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.lang.Nullable;
  7. import org.springframework.scheduling.annotation.AsyncConfigurer;
  8. import org.springframework.scheduling.annotation.EnableAsync;
  9. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  11. import java.util.concurrent.Executor;
  13. @EnableAsync
  14. public class MainConfig3 {
  16.     @Bean
  17.     public LogService logService() {
  18.         return new LogService();
  19.     }
  21.     /**
  22.      * 定义一个AsyncConfigurer类型的bean，实现getAsyncExecutor方法，返回自定义的线程池
  23.      *
  24.      * @param executor
  25.      * @return
  26.      */
  27.     @Bean
  28.     public AsyncConfigurer asyncConfigurer(@Qualifier("logExecutors") Executor executor) {
  29.         return new AsyncConfigurer() {
  30.             @Nullable
  31.             @Override
  32.             public Executor getAsyncExecutor() {
  33.                 return executor;
  34.             }
  35.         };
  36.     }
  38.     /**
  39.      * 定义一个线程池，用来异步处理日志方法调用
  40.      *
  41.      * @return
  42.      */
  43.     @Bean
  44.     public Executor logExecutors() {
  45.         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  46.         executor.setCorePoolSize(10);
  47.         executor.setMaxPoolSize(100);
  48.         //线程名称前缀
  49.         executor.setThreadNamePrefix("log-thread-"); //@1
  50.         return executor;
  51.     }
  53. }

@1自定义的线程池中线程名称前缀为log-thread-，运行下面测试代码

  1. @Test
  2. public void test3() throws InterruptedException {
  3.     AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
  4.     context.register(MainConfig3.class);
  5.     context.refresh();
  6.     LogService logService = context.getBean(LogService.class);
  7.     System.out.println(Thread.currentThread() + " logService.log start," + System.currentTimeMillis());
  8.     logService.log("异步执行方法!");
  9.     System.out.println(Thread.currentThread() + " logService.log end," + System.currentTimeMillis());
  11.     //休眠一下，防止@Test退出
  12.     TimeUnit.SECONDS.sleep(3);
  13. }

输出

  1. Thread[main,5,main] logService.log start,1595228732914
  2. Thread[main,5,main] logService.log end,1595228732921
  3. Thread[log-thread-1,5,main]开始记录日志,1595228732930
  4. Thread[log-thread-1,5,main]日志记录完毕,1595228734931

最后2行日志中线程名称是log-thread-，正是我们自定义线程池中的线程。

7、自定义异常处理

异步方法若发生了异常，我们如何获取异常信息呢？此时可以通过自定义异常处理来解决。

异常处理分2种情况

  1. 当返回值是Future的时候，方法内部有异常的时候，异常会向外抛出，可以对Future.get采用try..catch来捕获异常
  2. 当返回值不是Future的时候，可以自定义一个bean，实现AsyncConfigurer接口中的getAsyncUncaughtExceptionHandler方法，返回自定义的异常处理器

情况1：返回值为Future类型

用法

通过try..catch来捕获异常，如下

  1. try {
  2.     Future<String> future = logService.mockException();
  3.     System.out.println(future.get());
  4. } catch (ExecutionException e) {
  5.     System.out.println("捕获 ExecutionException 异常");
  6.     //通过e.getCause获取实际的异常信息
  7.     e.getCause().printStackTrace();
  8. } catch (InterruptedException e) {
  9.     e.printStackTrace();
  10. }

案例

LogService中添加一个方法，返回值为Future，内部抛出一个异常，如下：

  1. @Async
  2. public Future<String> mockException() {
  3.     //模拟抛出一个异常
  4.     throw new IllegalArgumentException("参数有误!");
  5. }

测试代码如下

  1. @Test
  2. public void test5() throws InterruptedException {
  3.     AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
  4.     context.register(MainConfig1.class);
  5.     context.refresh();
  6.     LogService logService = context.getBean(LogService.class);
  7.     try {
  8.         Future<String> future = logService.mockException();
  9.         System.out.println(future.get());
  10.     } catch (ExecutionException e) {
  11.         System.out.println("捕获 ExecutionException 异常");
  12.         //通过e.getCause获取实际的异常信息
  13.         e.getCause().printStackTrace();
  14.     } catch (InterruptedException e) {
  15.         e.printStackTrace();
  16.     }
  17.     //休眠一下，防止@Test退出
  18.     TimeUnit.SECONDS.sleep(3);
  19. }

运行输出

  1. java.lang.IllegalArgumentException: 参数有误!
  2. 捕获 ExecutionException 异常
  3.     at com.javacode2018.async.demo1.LogService.mockException(LogService.java:23)
  4.     at com.javacode2018.async.demo1.LogService$$FastClassBySpringCGLIB$$32a28430.invoke(<generated>)
  5.     at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218)

情况2：无返回值异常处理

用法

当返回值不是Future的时候，可以自定义一个bean，实现AsyncConfigurer接口中的getAsyncUncaughtExceptionHandler方法，返回自定义的异常处理器，当目标方法执行过程中抛出异常的时候，此时会自动回调AsyncUncaughtExceptionHandler#handleUncaughtException这个方法，可以在这个方法中处理异常，如下：

  1. @Bean
  2. public AsyncConfigurer asyncConfigurer() {
  3.     return new AsyncConfigurer() {
  4.         @Nullable
  5.         @Override
  6.         public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
  7.             return new AsyncUncaughtExceptionHandler() {
  8.                 @Override
  9.                 public void handleUncaughtException(Throwable ex, Method method, Object... params) {
  10.                     //当目标方法执行过程中抛出异常的时候，此时会自动回调这个方法，可以在这个方法中处理异常
  11.                 }
  12.             };
  13.         }
  14.     };
  15. }

案例

LogService中添加一个方法，内部抛出一个异常，如下：

  1. @Async
  2. public void mockNoReturnException() {
  3.     //模拟抛出一个异常
  4.     throw new IllegalArgumentException("无返回值的异常!");
  5. }

来个spring配置类，通过AsyncConfigurer来自定义异常处理器AsyncUncaughtExceptionHandler

  1. package com.javacode2018.async.demo4;
  3. import com.javacode2018.async.demo1.LogService;
  4. import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.lang.Nullable;
  7. import org.springframework.scheduling.annotation.AsyncConfigurer;
  8. import org.springframework.scheduling.annotation.EnableAsync;
  10. import java.lang.reflect.Method;
  11. import java.util.Arrays;
  13. @EnableAsync
  14. public class MainConfig4 {
  16.     @Bean
  17.     public LogService logService() {
  18.         return new LogService();
  19.     }
  21.     @Bean
  22.     public AsyncConfigurer asyncConfigurer() {
  23.         return new AsyncConfigurer() {
  24.             @Nullable
  25.             @Override
  26.             public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
  27.                 return new AsyncUncaughtExceptionHandler() {
  28.                     @Override
  29.                     public void handleUncaughtException(Throwable ex, Method method, Object... params) {
  30.                         String msg = String.format("方法[%s],参数[%s],发送异常了，异常详细信息:", method, Arrays.asList(params));
  31.                         System.out.println(msg);
  32.                         ex.printStackTrace();
  33.                     }
  34.                 };
  35.             }
  36.         };
  37.     }
  39. }

运行输出

  1. 方法[public void com.javacode2018.async.demo1.LogService.mockNoReturnException()],参数[[]],发送异常了，异常详细信息:
  2. java.lang.IllegalArgumentException: 无返回值的异常!
  3.     at com.javacode2018.async.demo1.LogService.mockNoReturnException(LogService.java:29)
  4.     at com.javacode2018.async.demo1.LogService$$FastClassBySpringCGLIB$$32a28430.invoke(<generated>)
  5.     at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218)

8、线程池隔离

什么是线程池隔离？

一个系统中可能有很多业务，比如充值服务、提现服务或者其他服务，这些服务中都有一些方法需要异步执行，默认情况下他们会使用同一个线程池去执行，如果有一个业务量比较大，占用了线程池中的大量线程，此时会导致其他业务的方法无法执行，那么我们可以采用线程隔离的方式，对不同的业务使用不同的线程池，相互隔离，互不影响。

@Async注解有个value参数，用来指定线程池的bean名称，方法运行的时候，就会采用指定的线程池来执行目标方法。

使用步骤

  1. 在spring容器中，自定义线程池相关的bean
  2. @Async(“线程池bean名称”)

案例

模拟2个业务：异步充值、异步提现；2个业务都采用独立的线程池来异步执行，互不影响。

异步充值服务
  1. package com.javacode2018.async.demo5;
  3. import org.springframework.scheduling.annotation.Async;
  4. import org.springframework.stereotype.Component;
  6. @Component
  7. public class RechargeService {
  8.     //模拟异步充值
  9.     @Async(MainConfig5.RECHARGE_EXECUTORS_BEAN_NAME)
  10.     public void recharge() {
  11.         System.out.println(Thread.currentThread() + "模拟异步充值");
  12.     }
  13. }
异步提现服务
  1. package com.javacode2018.async.demo5;
  3. import org.springframework.scheduling.annotation.Async;
  4. import org.springframework.stereotype.Component;
  6. @Component
  7. public class CashOutService {
  8.     //模拟异步提现
  9.     @Async(MainConfig5.CASHOUT_EXECUTORS_BEAN_NAME)
  10.     public void cashOut() {
  11.         System.out.println(Thread.currentThread() + "模拟异步提现");
  12.     }
  13. }
spring配置类

注意@0、@1、@2、@3、@4这几个地方的代码，采用线程池隔离的方式，注册了2个线程池，分别用来处理上面的2个异步业务。

  1. package com.javacode2018.async.demo5;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.ComponentScan;
  5. import org.springframework.scheduling.annotation.EnableAsync;
  6. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  8. import java.util.concurrent.Executor;
  10. @EnableAsync //@0：启用方法异步调用
  11. @ComponentScan
  12. public class MainConfig5 {
  14.     //@1：值业务线程池bean名称
  15.     public static final String RECHARGE_EXECUTORS_BEAN_NAME = "rechargeExecutors";
  16.     //@2：提现业务线程池bean名称
  17.     public static final String CASHOUT_EXECUTORS_BEAN_NAME = "cashOutExecutors";
  19.     /**
  20.      * @3：充值的线程池，线程名称以recharge-thread-开头
  21.      * @return
  22.      */
  23.     @Bean(RECHARGE_EXECUTORS_BEAN_NAME)
  24.     public Executor rechargeExecutors() {
  25.         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  26.         executor.setCorePoolSize(10);
  27.         executor.setMaxPoolSize(100);
  28.         //线程名称前缀
  29.         executor.setThreadNamePrefix("recharge-thread-");
  30.         return executor;
  31.     }
  33.     /**
  34.      * @4: 充值的线程池，线程名称以cashOut-thread-开头
  35.      *
  36.      * @return
  37.      */
  38.     @Bean(CASHOUT_EXECUTORS_BEAN_NAME)
  39.     public Executor cashOutExecutors() {
  40.         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  41.         executor.setCorePoolSize(10);
  42.         executor.setMaxPoolSize(100);
  43.         //线程名称前缀
  44.         executor.setThreadNamePrefix("cashOut-thread-");
  45.         return executor;
  46.     }
  47. }
测试代码
  1. @Test
  2. public void test7() throws InterruptedException {
  3.     AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
  4.     context.register(MainConfig5.class);
  5.     context.refresh();
  7.     RechargeService rechargeService = context.getBean(RechargeService.class);
  8.     rechargeService.recharge();
  9.     CashOutService cashOutService = context.getBean(CashOutService.class);
  10.     cashOutService.cashOut();
  12.     //休眠一下，防止@Test退出
  13.     TimeUnit.SECONDS.sleep(3);
  14. }
运行输出
  1. Thread[recharge-thread-1,5,main]模拟异步充值
  2. Thread[cashOut-thread-1,5,main]模拟异步提现

输出中可以看出2个业务使用的是不同的线程池执行的。

9、源码 & 原理

内部使用aop实现的，@EnableAsync会引入一个bean后置处理器：AsyncAnnotationBeanPostProcessor，将其注册到spring容器，这个bean后置处理器在所有bean创建过程中，判断bean的类上是否有@Async注解或者类中是否有@Async标注的方法，如果有，会通过aop给这个bean生成代理对象，会在代理对象中添加一个切面：org.springframework.scheduling.annotation.AsyncAnnotationAdvisor，这个切面中会引入一个拦截器：AnnotationAsyncExecutionInterceptor，方法异步调用的关键代码就是在这个拦截器的invoke方法中实现的，可以去看一下。

10、总结

11、案例源码

  1. https://gitee.com/javacode2018/spring-series

本博客所有系列案例代码以后都会放到这个上面，大家watch一下，可以持续关注动态。

最新资料

更多内容请访问：IT源点

注意：本文归作者所有，未经作者允许，不得转载

#Spring教程
阅读全部
相关文章推荐

全部评论: 0 条

    我有话说: