背景:
本文作为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 {
@Async
public 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取出业务逻辑类的bean
ComponentA 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 {
@Bean
public 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 {
@Async
public 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.PROXY
return 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;
}
}
// 类中是否注解了@Async
if (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倍)。