经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Spring Boot » 查看文章
Spring Boot 异步框架的使用详解
来源：jb51　　时间：2019/1/18 9:10:40　　对本文有异议

1. 前言

随着数据量和调用量的增长，用户对应用的性能要求越来越高。另外，在实际的服务中，还存在着这样的场景：系统在组装数据的时候，对于数据的各个部分的获取实际上是没有前后依赖关系的。这些问题都很容易让我们想到将这些同步调用全都改造为异步调用。不过自己实现起来比较麻烦，还容易出错。好在Spring已经提供了该问题的解决方案，而且使用起来十分方便。

2.Spring异步执行框架的使用方法

2.1 maven 依赖

Spring异步执行框架的相关bean包含在spring-context和spring-aop模块中，所以只要引入上述的模块即可。

2.2 开启异步任务支持

Spring提供了@EnableAsync的注解来标注是否启用异步任务支持。使用方式如下:

  1. @Configuration
  2. @EnableAsync
  3. public class AppConfig {
  4. }

Note: @EnableAsync必须要配合@Configuration使用，否则会不生效

2.3 方法标记为异步调用

将同步方法的调用改为异步调用也很简单。对于返回值为void的方法，直接加上@Async注解即可。对于有返回值的方法，除了加上上述的注解外，还需要将方法的返回值修改为Future类型和将返回值用AsyncResult包装起来。如下所示：

  1. // 无返回值的方法直接加上注解即可。 
  2. @Async
  3. public void method1() {
  4.  ...
  5. }
  6.  
  7. // 有返回值的方法需要修改返回值。
  8. @Async
  9. public Future<Object> method2() {
  10.  ...
  11.  return new AsyncResult<>(Object);
  12. }

2.4 方法调用

对于void的方法，和普通的调用没有任何区别。对于非void的方法，由于返回值是Future类型，所以需要用get()方法来获取返回值。如下所示：

  1. public static void main(String[] args) {
  2.   service.method1();
  3.   Future<Object> futureResult = service.method2();
  4.   Object result;
  5.   try {
  6.      result = futureResult.get(); 
  7.     } catch (InterruptedException | ExecutionException e) {
  8.       ...
  9.     }
  10. }

3. 原理简介

这块的源码的逻辑还是比较简单的，主要是Spring帮我们生成并管理了一个线程池，然后方法调用的时候使用动态代理将方法的执行包装为Callable类型并提交到线程池中执行。核心的实现逻辑在AsyncExecutionInterceptor类的invoke()方法中。如下所示：

  1. @Override
  2. public Object invoke(final MethodInvocation invocation) throws Throwable {
  3.   Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
  4.   Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
  5.   final Method userDeclaredMethod = BridgeMethodResolver.findBridgedMethod(specificMethod);
  6.  
  7.   AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod);
  8.   if (executor == null) {
  9.     throw new IllegalStateException(
  10.         "No executor specified and no default executor set on AsyncExecutionInterceptor either");
  11.   }
  12.  
  13.   Callable<Object> task = new Callable<Object>() {
  14.     @Override
  15.     public Object call() throws Exception {
  16.       try {
  17.         Object result = invocation.proceed();
  18.         if (result instanceof Future) {
  19.           return ((Future<?>) result).get();
  20.         }
  21.       }
  22.       catch (ExecutionException ex) {
  23.         handleError(ex.getCause(), userDeclaredMethod, invocation.getArguments());
  24.       }
  25.       catch (Throwable ex) {
  26.         handleError(ex, userDeclaredMethod, invocation.getArguments());
  27.       }
  28.       return null;
  29.     }
  30.   };
  31.  
  32.   return doSubmit(task, executor, invocation.getMethod().getReturnType());
  33. }

4.自定义taskExecutor及异常处理

4.1自定义taskExecutor

Spring查找TaskExecutor逻辑是：

1. 如果Spring context中存在唯一的TaskExecutor bean，那么就使用这个bean。

2. 如果1中的bean不存在，那么就会查找是否存在一个beanName为taskExecutor且是java.util.concurrent.Executor实例的bean，有则使用这个bean。

3. 如果1、2中的都不存在，那么Spring就会直接使用默认的Executor，即SimpleAsyncTaskExecutor。

在第2节的实例中，我们直接使用的是Spring默认的TaskExecutor。但是对于每一个新的任务，SimpleAysncTaskExecutor都是直接创建新的线程来执行，所以无法重用线程。具体的执行的代码如下：

  1. @Override
  2. public void execute(Runnable task, long startTimeout) {
  3.   Assert.notNull(task, "Runnable must not be null");
  4.   Runnable taskToUse = (this.taskDecorator != null ? this.taskDecorator.decorate(task) : task);
  5.   if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
  6.     this.concurrencyThrottle.beforeAccess();
  7.     doExecute(new ConcurrencyThrottlingRunnable(taskToUse));
  8.   }
  9.   else {
  10.     doExecute(taskToUse);
  11.   }
  12. }
  13.  
  14. protected void doExecute(Runnable task) {
  15.   Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task));
  16.   thread.start();
  17. }

所以我们在使用的时候，最好是使用自定义的TaskExecutor。结合上面描述的Spring查找TaskExecutor的逻辑，最简单的自定义的方法是使用@Bean注解。示例如下：

  1. // ThreadPoolTaskExecutor的配置基本等同于线程池
  2. @Bean("taskExecutor")
  3. public Executor getAsyncExecutor() {
  4.   ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
  5.   taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
  6.   taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
  7.   taskExecutor.setQueueCapacity(CORE_POOL_SIZE * 10);
  8.   taskExecutor.setThreadNamePrefix("wssys-async-task-thread-pool");
  9.   taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
  10.   taskExecutor.setAwaitTerminationSeconds(60 * 10);
  11.   taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
  12.   return taskExecutor;
  13. }

另外，Spring还提供了一个AsyncConfigurer接口，通过实现该接口，除了可以实现自定义Executor以外，还可以自定义异常的处理。代码如下：

  1. @Configuration
  2. @Slf4j
  3. public class AsyncConfig implements AsyncConfigurer {
  4.  
  5.   private static final int MAX_POOL_SIZE = 50;
  6.  
  7.   private static final int CORE_POOL_SIZE = 20;
  8.  
  9.   @Override
  10.   @Bean("taskExecutor")
  11.   public Executor getAsyncExecutor() {
  12.     ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
  13.  
  14.     taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
  15.     taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
  16.     taskExecutor.setQueueCapacity(CORE_POOL_SIZE * 10);
  17.     taskExecutor.setThreadNamePrefix("async-task-thread-pool");
  18.     taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
  19.     taskExecutor.setAwaitTerminationSeconds(60 * 10);
  20.     taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
  21.     return taskExecutor;
  22.   }
  23.  
  24.   @Override
  25.   public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
  26.     return (ex, method, params) -> log.error("invoke async method occurs error. method: {}, params: {}",
  27.       method.getName(), JSON.toJSONString(params), ex);
  28.   }
  29.  
  30. }

Note:

Spring还提供了一个AsyncConfigurerSupport类，该类也实现了AsyncConfigurer接口，且方法的返回值都是null，旨在提供一个方便的实现。

当getAsyncExecutor()方法返回null的时候，Spring会使用默认的处理器（强烈不推荐）。

当getAsyncUncaughtExceptionHandler()返回null的时候，Spring会使用SimpleAsyncUncaughtExceptionHandler来处理异常，该类会打印出异常的信息。

所以对该类的使用，最佳的实践是继承该类，并且覆盖实现getAsyncExecutor()方法。

4.2 异常处理

Spring异步框架对异常的处理如下所示：

  1. // 所在类：AsyncExecutionAspectSupport
  2. protected void handleError(Throwable ex, Method method, Object... params) throws Exception {
  3.   if (Future.class.isAssignableFrom(method.getReturnType())) {
  4.     ReflectionUtils.rethrowException(ex);
  5.   }
  6.   else {
  7.     // Could not transmit the exception to the caller with default executor
  8.     try {
  9.       this.exceptionHandler.handleUncaughtException(ex, method, params);
  10.     }
  11.     catch (Throwable ex2) {
  12.       logger.error("Exception handler for async method '" + method.toGenericString() +
  13.           "' threw unexpected exception itself", ex2);
  14.     }
  15.   }
  16. }

从代码来看，如果返回值是Future类型，那么直接将异常抛出。如果返回值不是Future类型（基本上包含的是所有返回值void类型的方法，因为如果方法有返回值，必须要用Future包装起来），那么会调用handleUncaughtException方法来处理异常。

注意：在handleUncaughtException()方法中抛出的任何异常，都会被Spring Catch住，所以没有办法将void的方法再次抛出并传播到上层调用方的！！！

关于Spring 这个设计的缘由我的理解是：既然方法的返回值是void，就说明调用方不关心方法执行是否成功，所以也就没有必要去处理方法抛出的异常。如果需要关心异步方法是否成功，那么返回值改为boolean就可以了。

4.4 最佳实践的建议

  1. @Async可以指定方法执行的Executor，用法：@Async("MyTaskExecutor")。推荐指定Executor，这样可以避免因为Executor配置没有生效而Spring使用默认的Executor的问题。

  2. 实现接口AsyncConfigurer的时候，方法getAsyncExecutor()必须要使用@Bean，并指定Bean的name。如果不使用@Bean，那么该方法返回的Executor并不会被Spring管理。用java doc api的原话是：is not a fully managed Spring bean.（具体含义没有太理解，不过亲测不加这个注解无法正常使用）

  3. 由于其本质上还是基于代理实现的，所以如果一个类中有A、B两个异步方法，而A中存在对B的调用，那么调用A方法的时候，B方法不会去异步执行的。

  4. 在异步方法上标注@Transactional是无效的。

  5. future.get()的时候，最好使用get(long timeout, TimeUnit unit)方法，避免长时间阻塞。

  6. ListenableFuture和CompletableFuture也是推荐使用的，他们相比Future，提供了对异步调用的各个阶段或过程进行介入的能力。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持w3xue。

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Spring Boot热门文章

  • 史上最详 Thymeleaf 使用教程
  • Spring Boot如何通过java -jar启动
  • Spring Boot 2.x（六）：优雅的统一返回值
  • Springboot使用Logback实现日志配置与异常记录
  • 使用SpringBoot+EasyExcel+Vue实现excel表格的导入和导出详解
  • spring boot上传文件超出大小异常无法捕获问题
  • paging组件的用法和意义
  • springboot2.0集成rabbitmq的示例代码
  • Spring MVC学习step1——框架熟悉
  • spring boot ${}占位符不起作用的解决方案

    • Spring Boot推荐文章

  • 3分钟带你搞定Spring Boot中Schedule
  • 不单独部署注册中心，又要具备注册中心的功能，我能上天！
  • 从零开始学Spring Boot系列-集成Spring Security实现用户认证与授权
  • 不单独部署注册中心，又要具备注册中心的功能，咋不让我上天？
  • EasyExcel 无法读取图片？用poi写了一个工具类
  • 【进阶篇】使用 Stream 流对比两个集合的常用操作分享
  • 关于DDD和COLA的一些总结和思考
  • SnakeYaml反序列化分析
  • RestTemplate进行https请求时适配信任证书
  • Kryo反序列化链分析

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号