Spring Boot异步框架的详细使用

1. 前言

随着数据量和调用量的增加，用户对应用程序性能的要求也越来越高. 另外，在实际的服务中，还存在这样的场景: 当系统组装数据时，实际上没有上下文相关性来获取数据的各个部分. 这些问题很容易让我们将所有这些同步调用视为异步调用. 但是，自行实施比较麻烦，而且容易出错. 幸运的是，Spring为这个问题提供了解决方案，并且使用起来非常方便.

2. 如何使用Spring异步执行框架

2.1 maven依赖

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

2.2启用异步任务支持

Spring提供@EnableAsync批注以标记是否启用异步任务支持. 用法如下:

@Configuration
@EnableAsync
public class AppConfig {
}

注意: @EnableAsync必须与@Configuration一起使用，否则不会生效

2.3方法标记为异步调用

将同步方法的调用更改为异步调用也很简单. 对于返回值为空的方法，只需添加@Async批注. 对于具有返回值的方法，除了上面的注释外，您还需要将方法的返回值修改为Future类型，并用AsyncResult包装返回值. 看起来像这样:

// 无返回值的方法直接加上注解即可。 
@Async
public void method1() {
 ...
}
// 有返回值的方法需要修改返回值。
@Async
public Future<Object> method2() {
 ...
 return new AsyncResult<>(Object);
}  

2.4方法调用

对于void方法，与普通调用没有区别. 对于非void方法，由于返回值的类型为Future，因此您需要使用get（）方法来获取返回值. 看起来像这样:

public static void main(String[] args) {
  service.method1();
  Future<Object> futureResult = service.method2();
  Object result;
  try {
     result = futureResult.get(); 
    } catch (InterruptedException | ExecutionException e) {
      ...
    }
}

3. 原理介绍

这部分源代码的逻辑相对简单，主要是因为Spring帮助我们生成和管理线程池，然后在调用该方法时，使用动态代理将方法的执行包装为可调用类型并将其提交到线程池以执行. . 核心实现逻辑在AsyncExecutionInterceptor类的invoke（）方法中. 看起来像这样:

@Override
public Object invoke(final MethodInvocation invocation) throws Throwable {
  Class<> targetClass = (invocation.getThis() != null  AopUtils.getTargetClass(invocation.getThis()) : null);
  Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
  final Method userDeclaredMethod = BridgeMethodResolver.findBridgedMethod(specificMethod);
  AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod);
  if (executor == null) {
    throw new IllegalStateException(
        "No executor specified and no default executor set on AsyncExecutionInterceptor either");
  }
  Callable<Object> task = new Callable<Object>() {
    @Override
    public Object call() throws Exception {
      try {
        Object result = invocation.proceed();
        if (result instanceof Future) {
          return ((Future<>) result).get();
        }
      }
      catch (ExecutionException ex) {
        handleError(ex.getCause(), userDeclaredMethod, invocation.getArguments());
      }
      catch (Throwable ex) {
        handleError(ex, userDeclaredMethod, invocation.getArguments());
      }
      return null;
    }
  };
  return doSubmit(task, executor, invocation.getMethod().getReturnType());
}

4. 自定义taskExecutor和异常处理

4.1自定义taskExecutor

Spring查找TaskExecutor的逻辑是:

1. 如果Spring上下文中有一个唯一的TaskExecutor bean，请使用此bean.

2. 如果1中的Bean不存在，则它将查找是否存在一个具有taskExecutor的taskName和java.util.concurrent.Executor实例的Bean. 如果有一个，请使用此bean.

3. 如果1和2都不存在，那么Spring将直接使用默认的Executor，即SimpleAsyncTaskExecutor.

在第2节的示例中，我们直接使用了Spring的默认TaskExecutor. 但是对于每个新任务，SimpleAysncTaskExecutor直接创建一个新线程来执行，因此该线程无法重用. 具体执行代码如下:

@Override
public void execute(Runnable task, long startTimeout) {
  Assert.notNull(task, "Runnable must not be null");
  Runnable taskToUse = (this.taskDecorator != null  this.taskDecorator.decorate(task) : task);
  if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
    this.concurrencyThrottle.beforeAccess();
    doExecute(new ConcurrencyThrottlingRunnable(taskToUse));
  }
  else {
    doExecute(taskToUse);
  }
}
protected void doExecute(Runnable task) {
  Thread thread = (this.threadFactory != null  this.threadFactory.newThread(task) : createThread(task));
  thread.start();
} 

因此，当我们使用它时，最好使用自定义TaskExecutor. 结合Spring查找上述TaskExecutor的逻辑，最简单的自定义方法是使用@Bean批注. 示例如下:

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

此外，Spring还提供了AsyncConfigurer接口. 通过实现此接口，除了自定义Executor外，您还可以自定义异常处理. 代码如下:

@Configuration
@Slf4j
public class AsyncConfig implements AsyncConfigurer {
  private static final int MAX_POOL_SIZE = 50;
  private static final int CORE_POOL_SIZE = 20;
  @Override
  @Bean("taskExecutor")
  public Executor getAsyncExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
    taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
    taskExecutor.setQueueCapacity(CORE_POOL_SIZE * 10);
    taskExecutor.setThreadNamePrefix("async-task-thread-pool");
    taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
    taskExecutor.setAwaitTerminationSeconds(60 * 10);
    taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
    return taskExecutor;
  }
  @Override
  public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
    return (ex, method, params) -> log.error("invoke async method occurs error. method: {}, params: {}",
      method.getName(), JSON.toJSONString(params), ex);
  }
}

本文来自电脑杂谈，转载请注明本文网址：
https://www.pc-fly.com/a/ruanjian/article-238381-1.html