异步任务spring @Async注解源码解析

/**
* Annotation that marks a method as a candidate for <i>asynchronous</i> execution.
* Can also be used at the type level, in which case all of the type's methods are
* considered as asynchronous.该注解可以标记一个异步执行的方法，也可以用来标注类，表示类中的所有方法都是异步执行的。
*
* <p>In terms of target method signatures, any parameter types are supported.
* However, the return type is constrained to either {@code void} or
* {@link java.util.concurrent.Future}. In the latter case, you may declare the
* more specific {@link org.springframework.util.concurrent.ListenableFuture} or
* {@link java.util.concurrent.CompletableFuture} types which allow for richer
* interaction with the asynchronous task and for immediate composition with
* further processing steps.入参随意，但返回值只能是void或者Future.(ListenableFuture接口/CompletableFuture类)
*
* <p>A {@code Future} handle returned from the proxy will be an actual asynchronous
* {@code Future} that can be used to track the result of the asynchronous method
* execution. However, since the target method needs to implement the same signature,
* it will have to return a temporary {@code Future} handle that just passes a value
* through: e.g. Spring's {@link AsyncResult}, EJB 3.1's {@link javax.ejb.AsyncResult},
* or {@link java.util.concurrent.CompletableFuture#completedFuture(Object)}.
* Future是代理返回的切实的异步返回，用以追踪异步方法的返回值。当然也可以使用AsyncResult类（实现ListenableFuture接口）(Spring或者EJB都有)或者CompletableFuture类
* @author Juergen Hoeller
* @author Chris Beams
* @since 3.0
* @see AnnotationAsyncExecutionInterceptor
* @see AsyncAnnotationAdvisor
*/
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface Async { /**
* A qualifier value for the specified asynchronous operation(s).
* <p>May be used to determine the target executor to be used when executing this
* method, matching the qualifier value (or the bean name) of a specific
* {@link java.util.concurrent.Executor Executor} or
* {@link org.springframework.core.task.TaskExecutor TaskExecutor}
* bean definition.用以限定执行方法的执行器名称（自定义）：Executor或者TaskExecutor
* <p>When specified on a class level {@code @Async} annotation, indicates that the
* given executor should be used for all methods within the class. Method level use
* of {@code Async#value} always overrides any value set at the class level.
* @since 3.1.2 加在类上表示整个类都使用，加在方法上会覆盖类上的设置
*/
String value() default ""; }

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(AsyncConfigurationSelector.class)
public @interface EnableAsync { /**该属性用来支持用户自定义异步注解，默认扫描spring的@Async和EJB3.1的@code @javax.ejb.Asynchronous
* Indicate the 'async' annotation type to be detected at either class
* or method level.
* <p>By default, both Spring's @{@link Async} annotation and the EJB 3.1
* {@code @javax.ejb.Asynchronous} annotation will be detected.
* <p>This attribute exists so that developers can provide their own
* custom annotation type to indicate that a method (or all methods of
* a given class) should be invoked asynchronously.
*/
Class<? extends Annotation> annotation() default Annotation.class; /**标明是否需要创建CGLIB子类代理，AdviceMode=PROXY时才适用。注意设置为true时，其它spring管理的bean也会升级到CGLIB子类代理
* Indicate whether subclass-based (CGLIB) proxies are to be created as opposed
* to standard Java interface-based proxies.
* <p><strong>Applicable only if the {@link #mode} is set to {@link AdviceMode#PROXY}</strong>.
* <p>The default is {@code false}.
* <p>Note that setting this attribute to {@code true} will affect <em>all</em>
* Spring-managed beans requiring proxying, not just those marked with {@code @Async}.
* For example, other beans marked with Spring's {@code @Transactional} annotation
* will be upgraded to subclass proxying at the same time. This approach has no
* negative impact in practice unless one is explicitly expecting one type of proxy
* vs. another &mdash; for example, in tests.
*/
boolean proxyTargetClass() default false; /**标明异步通知将会如何实现，默认PROXY，如需支持同一个类中非异步方法调用另一个异步方法，需要设置为ASPECTJ
* Indicate how async advice should be applied.
* <p><b>The default is {@link AdviceMode#PROXY}.</b>
* Please note that proxy mode allows for interception of calls through the proxy
* only. Local calls within the same class cannot get intercepted that way; an
* {@link Async} annotation on such a method within a local call will be ignored
* since Spring's interceptor does not even kick in for such a runtime scenario.
* For a more advanced mode of interception, consider switching this to
* {@link AdviceMode#ASPECTJ}.
*/
AdviceMode mode() default AdviceMode.PROXY; /**标明异步注解bean处理器应该遵循的执行顺序，默认最低的优先级（Integer.MAX_VALUE，值越小优先级越高）
* Indicate the order in which the {@link AsyncAnnotationBeanPostProcessor}
* should be applied.
* <p>The default is {@link Ordered#LOWEST_PRECEDENCE} in order to run
* after all other post-processors, so that it can add an advisor to
* existing proxies rather than double-proxy.
*/
int order() default Ordered.LOWEST_PRECEDENCE;}

/**查询器：基于@EanableAsync中定义的模式AdviceMode加在@Configuration标记的类上，确定抽象异步配置类的实现类
* Selects which implementation of {@link AbstractAsyncConfiguration} should be used based
* on the value of {@link EnableAsync#mode} on the importing {@code @Configuration} class.
*
* @author Chris Beams
* @since 3.1
* @see EnableAsync
* @see ProxyAsyncConfiguration
*/
public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> { private static final String ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME =
"org.springframework.scheduling.aspectj.AspectJAsyncConfiguration"; /**
* {@inheritDoc}
* @return {@link ProxyAsyncConfiguration} or {@code AspectJAsyncConfiguration} for
* {@code PROXY} and {@code ASPECTJ} values of {@link EnableAsync#mode()}, respectively
*/
@Override
public String[] selectImports(AdviceMode adviceMode) {
switch (adviceMode) {
case PROXY://如果配置的PROXY，使用ProxyAsyncConfiguration
return new String[] { ProxyAsyncConfiguration.class.getName() };
case ASPECTJ://如果配置的ASPECTJ，使用ProxyAsyncConfiguration
return new String[] { ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME };
default:
return null;
}
} }

/**
* {@code @Configuration} class that registers the Spring infrastructure beans necessary
* to enable proxy-based asynchronous method execution.
*
* @author Chris Beams
* @author Stephane Nicoll
* @since 3.1
* @see EnableAsync
* @see AsyncConfigurationSelector
*/
@Configuration
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public class ProxyAsyncConfiguration extends AbstractAsyncConfiguration { @Bean(name = TaskManagementConfigUtils.ASYNC_ANNOTATION_PROCESSOR_BEAN_NAME)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public AsyncAnnotationBeanPostProcessor asyncAdvisor() {
Assert.notNull(this.enableAsync, "@EnableAsync annotation metadata was not injected");
AsyncAnnotationBeanPostProcessor bpp = new AsyncAnnotationBeanPostProcessor();//新建一个异步注解bean后处理器
Class<? extends Annotation> customAsyncAnnotation = this.enableAsync.getClass("annotation");
//如果@EnableAsync中用户自定义了annotation属性，即异步注解类型，那么设置
if (customAsyncAnnotation != AnnotationUtils.getDefaultValue(EnableAsync.class, "annotation")) {
bpp.setAsyncAnnotationType(customAsyncAnnotation);
}
if (this.executor != null) {//Executor：设置线程任务执行器
bpp.setExecutor(this.executor);
}
if (this.exceptionHandler != null) {//AsyncUncaughtExceptionHandler:设置异常处理器
bpp.setExceptionHandler(this.exceptionHandler);
}
bpp.setProxyTargetClass(this.enableAsync.getBoolean("proxyTargetClass"));//设置是否升级到CGLIB子类代理，默认不开启
bpp.setOrder(this.enableAsync.<Integer>getNumber("order"));//设置执行优先级，默认最后执行
return bpp;
} }

1 /**
2 * Abstract base {@code Configuration} class providing common structure for enabling
3 * Spring's asynchronous method execution capability.
4 * 抽象异步配置类，封装了通用结构，用以支持spring的异步方法执行能力
5 * @author Chris Beams
6 * @author Stephane Nicoll
7 * @since 3.1
8 * @see EnableAsync
9 */
10 @Configuration
11 public abstract class AbstractAsyncConfiguration implements ImportAware {
12
13 protected AnnotationAttributes enableAsync;//enableAsync的注解属性 protected Executor executor;//Doug Lea老李头设计的线程任务执行器
16
protected AsyncUncaughtExceptionHandler exceptionHandler;//异常处理器 @Override
public void setImportMetadata(AnnotationMetadata importMetadata) {
this.enableAsync = AnnotationAttributes.fromMap(
importMetadata.getAnnotationAttributes(EnableAsync.class.getName(), false));
if (this.enableAsync == null) {
throw new IllegalArgumentException(
"@EnableAsync is not present on importing class " + importMetadata.getClassName());
}
} /**
* Collect any {@link AsyncConfigurer} beans through autowiring.
*/
@Autowired(required = false)
void setConfigurers(Collection<AsyncConfigurer> configurers) {
if (CollectionUtils.isEmpty(configurers)) {
return;
}
if (configurers.size() > 1) {
throw new IllegalStateException("Only one AsyncConfigurer may exist");
}
AsyncConfigurer configurer = configurers.iterator().next();
this.executor = configurer.getAsyncExecutor();
this.exceptionHandler = configurer.getAsyncUncaughtExceptionHandler();
} }

1 @Override
2 public void setBeanFactory(BeanFactory beanFactory) {
3 super.setBeanFactory(beanFactory);
4
5 AsyncAnnotationAdvisor advisor = new AsyncAnnotationAdvisor(this.executor, this.exceptionHandler);
6 if (this.asyncAnnotationType != null) {
7 advisor.setAsyncAnnotationType(this.asyncAnnotationType);
8 }
9 advisor.setBeanFactory(beanFactory);
10 this.advisor = advisor;
11 }

@Override
public Object postProcessAfterInitialization(Object bean, String beanName) {
if (bean instanceof AopInfrastructureBean) {
// Ignore AOP infrastructure such as scoped proxies.
return bean;
}
//把Advisor添加进bean ProxyFactory-》AdvisedSupport-》Advised
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;
}
}
//构造ProxyFactory代理工厂，添加代理的接口，设置切面，最后返回代理类：AopProxy
if (isEligible(bean, beanName)) {
ProxyFactory proxyFactory = prepareProxyFactory(bean, beanName);
if (!proxyFactory.isProxyTargetClass()) {
evaluateProxyInterfaces(bean.getClass(), proxyFactory);
}
proxyFactory.addAdvisor(this.advisor);
customizeProxyFactory(proxyFactory);
return proxyFactory.getProxy(getProxyClassLoader());
} // No async proxy needed.
return bean;
}

public static boolean canApply(Advisor advisor, Class<?> targetClass, boolean hasIntroductions) {
if (advisor instanceof IntroductionAdvisor) {
return ((IntroductionAdvisor) advisor).getClassFilter().matches(targetClass);
}//满足第二分支PointcutAdvisor
else if (advisor instanceof PointcutAdvisor) {
PointcutAdvisor pca = (PointcutAdvisor) advisor;
return canApply(pca.getPointcut(), targetClass, hasIntroductions);
}
else {
// It doesn't have a pointcut so we assume it applies.
return true;
}
}

// 得到方法的拦截器链
List<Object> chain = this.advised.getInterceptorsAndDynamicInterceptionAdvice(method, targetClass);
// Check whether we have any advice. If we don't, we can fallback on direct
// reflective invocation of the target, and avoid creating a MethodInvocation.
if (chain.isEmpty()) {
// We can skip creating a MethodInvocation: just invoke the target directly
// Note that the final invoker must be an InvokerInterceptor so we know it does
// nothing but a reflective operation on the target, and no hot swapping or fancy proxying.
Object[] argsToUse = AopProxyUtils.adaptArgumentsIfNecessary(method, args);
retVal = AopUtils.invokeJoinpointUsingReflection(target, method, argsToUse);
}
else {
// 构造
invocation = new ReflectiveMethodInvocation(proxy, target, method, args, targetClass, chain);
// Proceed to the joinpoint through the interceptor chain.
retVal = invocation.proceed();
}

@Override
public Object proceed() throws Throwable {
// 如果没有拦截器，直接执行被代理的方法
if (this.currentInterceptorIndex == this.interceptorsAndDynamicMethodMatchers.size() - 1) {
return invokeJoinpoint();
} Object interceptorOrInterceptionAdvice =
this.interceptorsAndDynamicMethodMatchers.get(++this.currentInterceptorIndex);
if (interceptorOrInterceptionAdvice instanceof InterceptorAndDynamicMethodMatcher) {
// Evaluate dynamic method matcher here: static part will already have
// been evaluated and found to match.
InterceptorAndDynamicMethodMatcher dm =
(InterceptorAndDynamicMethodMatcher) interceptorOrInterceptionAdvice;
if (dm.methodMatcher.matches(this.method, this.targetClass, this.arguments)) {
return dm.interceptor.invoke(this);
}
else {
// Dynamic matching failed.
// Skip this interceptor and invoke the next in the chain.
return proceed();
}
}
else {
// It's an interceptor, so we just invoke it: The pointcut will have
// been evaluated statically before this object was constructed.
return ((MethodInterceptor) interceptorOrInterceptionAdvice).invoke(this);
}
}

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());
}

protected Object doSubmit(Callable<Object> task, AsyncTaskExecutor executor, Class<?> returnType) {
if (completableFuturePresent) {//先判断是否存在CompletableFuture这个类，优先使用CompletableFuture执行任务
Future<Object> result = CompletableFutureDelegate.processCompletableFuture(returnType, task, executor);
if (result != null) {
return result;
}
}//返回值是可监听Future，定义过回调函数：addCallback
8 if (ListenableFuture.class.isAssignableFrom(returnType)) {
9 return ((AsyncListenableTaskExecutor) executor).submitListenable(task);
10 }//返回值是Future
11 else if (Future.class.isAssignableFrom(returnType)) {
12 return executor.submit(task);
13 }
14 else {//没有返回值
15 executor.submit(task);
16 return null;
17 }
18 }

protected void doExecute(Runnable task) {
Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task));
thread.start();
}