Spring系列-9 Async注解使用与原理

背景：

本文作为Spring系列的第九篇，介绍@Async注解的使用、注意事项和实现原理，原理部分会结合Spring框架代码进行。

本文可以和Spring系列-8 AOP原理进行比较阅读

1.使用方式

@Async一般注解在方法上，用于实现方法的异步：方法调用者立即返回，待调用的方法提交给Spring的线程池执行。@Async也可以注解在类上，等价于在类中的所有方法上添加该注解。需要注意@Async只对Spring管理的对象生效。

案例介绍：

1.1 基本用例

在配置类或者启动类上注解@EnableAsync，用于开启异步功能：

@Configuration
@ComponentScan(basePackages = "com.seong.async")
@EnableAsync
public class AsyncConfiguration {
}

在方法上添加@Async注解：

package com.seong.async;@Component
public class ComponentA {@Asyncpublic void print() {System.out.println(Thread.currentThread().getName()+":"+" test Async call.");}
}

用例如下：

@Slf4j
public class AsyncApplication {public static void main(String[] args) {AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext(AsyncConfiguration.class);//用id取出业务逻辑类的beanComponentA componentA = (ComponentA) applicationContext.getBean("componentA");LOGGER.info("test begin...");componentA.print();LOGGER.info("test end!");}
}

得到如下结果：

结果显示：ComponentA对象的print()方法被异步执行了，且DEBUG日志抛出了No qualifying bean of type 'org.springframework.core.task.TaskExecutor' available异常；这是因为业务代码没有配置TaskExecutor类型的Bean对象导致。

1.2 配置线程池

被@Async注解的方法会提交给线程池执行，这里可以手动指定线程池或者使用默认的线程池:
(1) 手动指定线程池：
通过Async的value属性指定线程池Bean对象(通过beanName指定)，如：

@Component
public class ComponentA {@Async("myTaskExecutor")public void print() {System.out.println(Thread.currentThread().getName()+":"+" test Async call.");}
}@Configuration
@EnableAsync
public class AsyncConfiguration {@Beanpublic Executor myTaskExecutor() {return Executors.newFixedThreadPool(1);}
}

此时，运行任务会提交给"myTaskExecutor"线程池对象执行。

(2) 配置默认的线程池：

@Bean
public TaskExecutor myTaskExecutor() {return new SimpleAsyncTaskExecutor();
}

Spring为@Async提供了默认线程池配置，可通过向IOC中注册TaskExecutor类型的Bean对象实现。
也可使用如下配置注册默认线程池：

@Bean
public Executor taskExecutor() {return Executors.newFixedThreadPool(1);
}

Spring框架获取TaskExecutor类型的Bean对象失败时，会尝试获取BeanName为"taskExecutor"的线程池对象；但执行时日志中会给出异常信息。

(3) 使用Spring框架默认的SimpleAsyncTaskExecutor线程池.
若业务未配置默认线程池，默认使用Spring生成的SimpleAsyncTaskExecutor对象； 但执行时日志中会给出异常信息。

1.3 获取返回值

由于@Async注解的方法为异步执行，因此可以通过Future来获取返回值，案例如下：
修改ComponentA的方法：

@Component
@Slf4j
public class ComponentA {@Asyncpublic Future<String> print() {LOGGER.info(Thread.currentThread().getName() + ":" + " test Async call.");return new AsyncResult<>("ComponentA finished.");}
}

适配修改案例：

@Slf4j
public class AsyncApplication {public static void main(String[] args) throws ExecutionException, InterruptedException {AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext(AsyncConfiguration.class);ComponentA componentA = (ComponentA) applicationContext.getBean("componentA");LOGGER.info("test begin...");final Future<String> resultFuture = componentA.print();doOtherBusiness();LOGGER.info("test end!");// 这一步执行时main线程会等待异步方法返回的结果LOGGER.info("result is {}", resultFuture.get());}private static void doOtherBusiness() {LOGGER.info("do other business...");}
}

运行得到如下结果：

2.原理：

本质上Async注解的实现依赖于动态代理，代理过程中将任务提交给线程池，存在以下流程：

如上图所示，Spring为使用Async注解Bean对象生产一个动态代理对象，当Async注解的方法被调用时，会进入代理流程，选择线程池、封装调用逻辑并提交给线程池执行，返回执行结果。
注意：未被@Async注解的方法则不会执行上述流程(static方法也不会)
由于异步的本质是基于代理实现，所以同一个类中的方法调用会导致被调用方的异步作用失效，该场景与Spring的事务失效原因相同，可参考：事务-1 事务隔离级别和Spring事务传播机制.

2.1 EnableAsync注解：

2.1.1 EnableAsync注解定义

@EnableAsync注解核心作用是向容器中注册AsyncAnnotationBeanPostProcessor对象。

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(AsyncConfigurationSelector.class)
public @interface EnableAsync {Class<? extends Annotation> annotation() default Annotation.class;boolean proxyTargetClass() default false;AdviceMode mode() default AdviceMode.PROXY;int order() default Ordered.LOWEST_PRECEDENCE;
}

@EnableAsync注解提过了以下属性用于自定义配置：
[1] annotation属性：
用于指定生效的注解，默认为@Async和EJB的@Asynchronous注解。当指定注解时，仅指定的注解生效：

@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface TestAnnotation {
}@EnableAsync(annotation = TestAnnotation.class)
public class AsyncConfiguration { 
}@TestAnnotation
public void print() {System.out.println(Thread.currentThread().getName() + ":" + " test Async call");
}

但是，仅@Async中可以通过value属性指定需要的线程池Bean对象；因为硬编码获取beanName的逻辑：来源必须为@Async注解的value属性：

protected String getExecutorQualifier(Method method) {Async async = AnnotatedElementUtils.findMergedAnnotation(method, Async.class);if (async == null) {async = AnnotatedElementUtils.findMergedAnnotation(method.getDeclaringClass(), Async.class);}return (async != null ? async.value() : null);
}

[2] proxyTargetClass属性：
可用于配置代理类型：true时表示强制使用CGLIB动态代理，false时表示根据被代理类情况进行确定，默认为false。仅当模式设置为AdviceMode#PROXY时有效。

[3] mode属性：
指出应该如何应用异步通知, 默认值是AdviceMode.PROXY。

[4] order属性：
指示应该Bean对象在BPP阶段应用AsyncAnnotationBeanPostProcessor的顺序，默认值是Ordered.LOWEST_PRECEDENCE，以便在所有其他后处理器之后运行，这样它就可以向现有代理添加一个advisor而不是双代理。

2.1.2 EnableAsync注解作用

通过@Import注解向IOC中注入了AsyncConfigurationSelector对象, 进入AsyncConfigurationSelector对象：

public final String[] selectImports(AnnotationMetadata importingClassMetadata) {//...AdviceMode adviceMode = attributes.getEnum(getAdviceModeAttributeName());String[] imports = selectImports(adviceMode);//...return imports;
}public String[] selectImports(AdviceMode adviceMode) {switch (adviceMode) {case PROXY:// @EnableAsync注解的mode属性的默认值为AdviceMode.PROXYreturn new String[] {ProxyAsyncConfiguration.class.getName()};case ASPECTJ:return new String[] {ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME};default:return null;}
}

由于加载的AsyncConfigurationSelector对象为ImportSelector类型，Spring继续完成ProxyAsyncConfiguration类型的注入。
进入ProxyAsyncConfiguration代码逻辑，发现是一个配置类，用于向容器中注入一个AsyncAnnotationBeanPostProcessor对象：

@Configuration
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public class ProxyAsyncConfiguration extends AbstractAsyncConfiguration {@Bean(name="org.springframework.context.annotation.internalAsyncAnnotationProcessor")@Role(BeanDefinition.ROLE_INFRASTRUCTURE)public AsyncAnnotationBeanPostProcessor asyncAdvisor() {AsyncAnnotationBeanPostProcessor bpp = new AsyncAnnotationBeanPostProcessor();bpp.configure(this.executor, this.exceptionHandler);Class<? extends Annotation> customAsyncAnnotation = this.enableAsync.getClass("annotation");if (customAsyncAnnotation != AnnotationUtils.getDefaultValue(EnableAsync.class, "annotation")) {bpp.setAsyncAnnotationType(customAsyncAnnotation);}bpp.setProxyTargetClass(this.enableAsync.getBoolean("proxyTargetClass"));bpp.setOrder(this.enableAsync.<Integer>getNumber("order"));return bpp;}
}

总之，@EnableAsync后IOC容器中会增加一个AsyncAnnotationBeanPostProcessor类型的对象；beanName为"org.springframework.context.annotation.internalAsyncAnnotationProcessor".

2.2 AsyncAnnotationBeanPostProcessor:

AsyncAnnotationBeanPostProcessor是一个后置处理器且实现了BeanFactoryAware接口，核心逻辑在两处：Aware阶段和BPP-After阶段.
[1] Aware阶段：准备好Advisor对象(内部包含一个拦截器)

public void setBeanFactory(BeanFactory beanFactory) {super.setBeanFactory(beanFactory);// 构造AsyncAnnotationAdvisor对象，注入到this.advisor属性中AsyncAnnotationAdvisor advisor = new AsyncAnnotationAdvisor(this.executor, this.exceptionHandler);advisor.setBeanFactory(beanFactory);this.advisor = advisor;
}

上述方法完成了BeanFactory属性的注入以及this.advisor属性的设置，属性类型为AsyncAnnotationAdvisor，进入AsyncAnnotationAdvisor的构造函数：

public AsyncAnnotationAdvisor( executor, exceptionHandler) {this.advice = buildAdvice(executor, exceptionHandler);this.pointcut = buildPointcut(asyncAnnotationTypes);
}protected Advice buildAdvice( executor, exceptionHandler) {AnnotationAsyncExecutionInterceptor interceptor = new AnnotationAsyncExecutionInterceptor(null);// 设置线程池和异常处理器interceptor.configure(executor, exceptionHandler);return interceptor;
}

注意AnnotationAsyncExecutionInterceptor是一个MethodInterceptor接口的实现类：

// MethodInterceptor拦截器接口的invoke方法
public Object invoke(final MethodInvocation invocation) throws Throwable {//...
}

总之，在aware阶段为AsyncAnnotationBeanPostProcessor对象注入this.advisor一个AsyncAnnotationAdvisor类型的对象，该对象包含BeanFactory和advice属性；其中advice属性包含一个MethodInterceptor拦截器。

[2] BPP-After阶段：为Bean对象生成代理
在AsyncAnnotationBeanPostProcessor的初始化过程完成后，当IOC初始化单例Bean对象时，会在初始化的后置处理器阶段调用AsyncAnnotationBeanPostProcessor的postProcessAfterInitialization方法：

@Override
public Object postProcessAfterInitialization(Object bean, String beanName) {// this.advisor在AsyncAnnotationBeanPostProcessor构建阶段已设值if (this.advisor == null || bean instanceof AopInfrastructureBean) {return bean;}// 是否已进行了AOP代理if (bean instanceof Advised) {Advised advised = (Advised) bean;if (!advised.isFrozen() && isEligible(AopUtils.getTargetClass(bean))) {// Add our local Advisor to the existing proxy's Advisor chain...if (this.beforeExistingAdvisors) {advised.addAdvisor(0, this.advisor);}else {advised.addAdvisor(this.advisor);}return bean;}}// 类中是否注解了@Asyncif (isEligible(bean, beanName)) {// 将bean对象存在了targetSource属性中，以便后续取用ProxyFactory proxyFactory = prepareProxyFactory(bean, beanName);if (!proxyFactory.isProxyTargetClass()) {evaluateProxyInterfaces(bean.getClass(), proxyFactory);}proxyFactory.addAdvisor(this.advisor);customizeProxyFactory(proxyFactory);return proxyFactory.getProxy(getProxyClassLoader());}return bean;
}

如上述代码所示：
已完成AOP代理的不需要再次代理，将AsyncAnnotationBeanPostProcessor的this.advisor属性添加到代理对象的List<Advisor>列表中；未经过代理且类中注解了@Async的对象，Spring为其创建一个代理对象，并将this.advisor属性信息保存到代理对象中。【代理过程参考：Spring系列-8 AOP原理】

【3】接口调用
Spring提供了两种代理方式：JDK动态代理和CGLIB代理，由于底层代理细节对实现原理并无影响，这里以JDK动态代理为例进行介绍。
被代理的对象在方法被调用时，进入JdkDynamicAopProxy的invoke拦截方法，注意每个代理对象对应一个JdkDynamicAopProxy对象。

[插图：解释ProxyFactory、ProxyConfig、Advised的关系，以及JdkDynamicAopProxy的传参路径]

涉及代码如下所示(保留主线逻辑)：

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {// hash/equals/toString check...TargetSource targetSource = this.advised.targetSource;Object target = targetSource.getTarget();Class<?> targetClass = (target != null ? target.getClass() : null);// 根据方法和字节码对象获取调用链，如果该方法未被@Async注解，则返回的集合为空；否则返回一个包含AnnotationAsyncExecutionInterceptor对象的单元素集合。List<Object> chain = this.advised.getInterceptorsAndDynamicInterceptionAdvice(method, targetClass);Object retVal;if (chain.isEmpty()) {Object[] argsToUse = AopProxyUtils.adaptArgumentsIfNecessary(method, args);// 直接用过反射调用目标方法retVal = AopUtils.invokeJoinpointUsingReflection(target, method, argsToUse);} else {MethodInvocation invocation =new ReflectiveMethodInvocation(proxy, target, method, args, targetClass, chain);retVal = invocation.proceed();}// 返回结果类型校验...return retVal;
}

invoke方法的主线逻辑：
[1] 从targetSource属性中取出被代理的对象;
[2] 从 advised属性中获取拦截器链;
[3] 判断拦截器链是否为空，为空则直接通过反射调用该方法；否则进入拦截器对象；
[4] 进行结果类型的校验，并返回结果对象。

ReflectiveMethodInvocation介绍：

ReflectiveMethodInvocation是对调用链进行的一层封装：

public Object proceed() throws Throwable {// We start with an index of -1 and increment early.if (this.currentInterceptorIndex == this.interceptorsAndDynamicMethodMatchers.size() - 1) {return invokeJoinpoint();}Object interceptorOrInterceptionAdvice =this.interceptorsAndDynamicMethodMatchers.get(++this.currentInterceptorIndex);return ((MethodInterceptor) interceptorOrInterceptionAdvice).invoke(this);
}

interceptorsAndDynamicMethodMatchers为构建ReflectiveMethodInvocation对象时传入的chain参数；即依次调用拦截器的拦截invoke方法，最后执行invokeJoinpoint()方法。
注意：调用拦截器时将this作为入参(该ReflectiveMethodInvocation对象)，拦截器中可通过调用该对象的proceed()实现回调。

2.3 AnnotationAsyncExecutionInterceptor介绍：

当异步方法被调用时，堆栈逻辑进入到AnnotationAsyncExecutionInterceptor的invoke方法：

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);// 1.确定待执行的任务的线程池AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod);// 2.封装执行逻辑成一个任务Callable<Object> task = () -> {try {// 回调ReflectiveMethodInvocation的proceed方法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;};// 3.将任务提交给线程池return doSubmit(task, executor, invocation.getMethod().getReturnType());
}

AnnotationAsyncExecutionInterceptor的invoke方法主线逻辑如下：
[1] 获取待执行的任务的线程池;
[2] 回调ReflectiveMethodInvocation的逻辑(将invocation.proceed())封装为一个任务；
[3] 将任务提交给线程池执行。

当提交给线程池的任务被执行时，即invocation.proceed()执行时进入到ReflectiveMethodInvocation的invokeJoinpoint()方法，反射调用目标方法。

获取执行@Async方法的线程池：

protected AsyncTaskExecutor determineAsyncExecutor(Method method) {AsyncTaskExecutor executor = this.executors.get(method);if (executor == null) {Executor targetExecutor;String qualifier = getExecutorQualifier(method);if (StringUtils.hasLength(qualifier)) {targetExecutor = findQualifiedExecutor(this.beanFactory, qualifier);}else {targetExecutor = this.defaultExecutor.get();}if (targetExecutor == null) {return null;}executor = (targetExecutor instanceof AsyncListenableTaskExecutor ?(AsyncListenableTaskExecutor) targetExecutor : new TaskExecutorAdapter(targetExecutor));this.executors.put(method, executor);}return executor;
}

其中，this.executors对象用于缓存方法与线程池的关系，减少查询次数。

主线逻辑为：
如果@Async注解的value有值，则根据该值从beanFactory中获取对应的线程池Bean对象；否则获取默认的线程池，获取步骤如下：

protected Executor getDefaultExecutor(@Nullable BeanFactory beanFactory) {Executor defaultExecutor = super.getDefaultExecutor(beanFactory);return (defaultExecutor != null ? defaultExecutor : new SimpleAsyncTaskExecutor());
}

逻辑较为简单：通过父类的getDefaultExecutor方法获取默认的线程池，获取失败则使用SimpleAsyncTaskExecutor线程池对象。

父类getDefaultExecutor方法逻辑如下：

protected Executor getDefaultExecutor(@Nullable BeanFactory beanFactory) {if (beanFactory != null) {try {return beanFactory.getBean(TaskExecutor.class);} catch (NoUniqueBeanDefinitionException ex) {return beanFactory.getBean("taskExecutor", Executor.class);} catch (NoSuchBeanDefinitionException ex) {return beanFactory.getBean("taskExecutor", Executor.class);}}return null;
}

先尝试根据TaskExecutor类型从IOC中获取Bean对象，获取失败再次根据"taskExecutor"名称获取Executor类型的Bean对象，都获取失败时返回null；当从父类的getDefaultExecutor方法获取线程池结果为空时，会使用 SimpleAsyncTaskExecutor线程池对象。

注意：
SimpleAsyncTaskExecutor对象每次执行任务都会创建一个新的线程(且没有最大线程数设置)，当并发量较大时可能导致严重的性能问题；建议使用@Async的项目自定义beanName为 "taskExecutor"且类型为TaskExecutor的线程池。

2.4 代理类型：

Spring提供了两种动态代理类型：JDK动态代理和CGLIB动态代理(可参考：AVASE-14 静态代理与动态代理)，并支持通过配置对其进行指定。
AopProxy工厂(DefaultAopProxyFactory)创建AopProxy对象的过程如下：

@Override
public AopProxy createAopProxy(AdvisedSupport config) throws AopConfigException {if (!NativeDetector.inNativeImage() &&(config.isOptimize() || config.isProxyTargetClass() || hasNoUserSuppliedProxyInterfaces(config))) {Class<?> targetClass = config.getTargetClass();if (targetClass == null) {throw new AopConfigException("TargetSource cannot determine target class: " +"Either an interface or a target is required for proxy creation.");}if (targetClass.isInterface() || Proxy.isProxyClass(targetClass)) {return new JdkDynamicAopProxy(config);}return new ObjenesisCglibAopProxy(config);} else {return new JdkDynamicAopProxy(config);}
}

NativeDetector.inNativeImage()和config.isOptimize() 在整个代理和方法调用过程(未涉及配置)取默认值，可直接忽略；且通过Class<?> targetClass = config.getTargetClass();得到的对象为(已被实例化的)Bean对象的字节码类型，不可能为接口类型或空对象，可直接忽略。[框架代码不同于业务代码，需要考虑多种场景的适配]
因此，上述代码可以简化为：

@Override
public AopProxy createAopProxy(AdvisedSupport config) throws AopConfigException {if (config.isProxyTargetClass() || hasNoUserSuppliedProxyInterfaces(config)) {Class<?> targetClass = config.getTargetClass();if (Proxy.isProxyClass(targetClass)) {return new JdkDynamicAopProxy(config);}return new ObjenesisCglibAopProxy(config);} else {return new JdkDynamicAopProxy(config);}
}// 目标类是否继承了接口（除SpringProxy外）
private boolean hasNoUserSuppliedProxyInterfaces(AdvisedSupport config) {Class<?>[] ifcs = config.getProxiedInterfaces();return (ifcs.length == 0 || (ifcs.length == 1 && SpringProxy.class.isAssignableFrom(ifcs[0])));
}

结果比较清晰：
【1】如果目标对象已被JDK动态代理过，则选择JDK动态代理；
【2】如果ProxyTarget属性为true 或者没有实现接口，使用CGBLIB代理；否则使用JDK动态代理。

顺便提一下：JDK动态代理生成代理类的速度较快(相对CGLLIB快8倍)，但是运行速度较慢(比CGLIB慢10倍)。