之前看了一篇文章，里面提到了使用@Async注解的两个问题，第一个是Spring中实现@Async注解的线程池的阻塞队列是无界队列LinkedBlockingQueue，这就导致最大线程数的配置是无效的，如果异步任务很多且执行时间较长，会导致任务一直堆积在队列中，任务延迟很大。第二个问题是在SpringBoot中，如果没有自定义线程池实例，那么SpringBoot会使用默认的线程池，这个默认线程池是SimpleAsyncTaskExecutor，这种线程池是会为每个任务创建一个线程去执行，可能会引起资源问题。
因为项目中也用到了@Async注解，为了了解@Async的原理，决定从头开始撸一遍SpringBoot中的@Async注解的源码。
应用启动阶段
实例化AsyncConfigurationSelector
先看@EnableAsync注解
@Target(ElementType.TYPE)@Retention(RetentionPolicy.RUNTIME)@Documented@Import(AsyncConfigurationSelector.class)public @interface EnableAsync { /** * 自定义异步注解，@Async和@javax.ejb.Asynchronous默认是会被检测到的 */ Class<? extends Annotation> annotation() default Annotation.class; /** * 表示是否使用子类代理（CGLIB）还是基于接口的代理（JDK代理） */ boolean proxyTargetClass() default false; /** * 表示使用哪种advice，PROXY是基于代理的，另外一种是切面织入形式的 */ AdviceMode mode() default AdviceMode.PROXY; /** * 表示AsyncAnnotationBeanPostProcessor这个后置处理器的应用顺序 */ int order() default Ordered.LOWEST_PRECEDENCE;}@Import注解导入了AsyncConfigurationSelector类，继承自AdviceModeImportSelector
public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> { private static final String ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME = "org.springframework.scheduling.aspectj.AspectJAsyncConfiguration"; /** * 根据@EnableAsync的mode属性返回不同配置类 */ @Override public String[] selectImports(AdviceMode adviceMode) { switch (adviceMode) { case PROXY: return new String[] { ProxyAsyncConfiguration.class.getName() }; case ASPECTJ: return new String[] { ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME }; default: return null; } }}/** * 基于注解的mode属性来获取imports的基类 */public abstract class AdviceModeImportSelector<A extends Annotation> implements ImportSelector { public static final String DEFAULT_ADVICE_MODE_ATTRIBUTE_NAME = "mode"; protected String getAdviceModeAttributeName() { return DEFAULT_ADVICE_MODE_ATTRIBUTE_NAME; } /** * 获取泛型注解的mode属性，调用子类的selectImports(AdviceMode adviceMode)方法获取import配置类 * AnnotationMetadata importingClassMetadata是SpringBoot启动类的上获取的注解（我当前项目上就@EnableAsync, @SpringBootApplication两个注解）, ConfigurationClassParser的processImports方法传进来的 */ @Override public final String[] selectImports(AnnotationMetadata importingClassMetadata) { //获取当前类的泛型参数（我自己Debug时就是@EnableAsync, getClass()获取到的是AsyncConfigurationSelector的class对象） Class<?> annoType = GenericTypeResolver.resolveTypeArgument(getClass(), AdviceModeImportSelector.class); //获取指定当前泛型注解属性和值 AnnotationAttributes attributes = AnnotationConfigUtils.attributesFor(importingClassMetadata, annoType); if (attributes == null) { throw new IllegalArgumentException(String.format( "@%s is not present on importing class '%s' as expected", annoType.getSimpleName(), importingClassMetadata.getClassName())); } //获取mode属性值 AdviceMode adviceMode = attributes.getEnum(this.getAdviceModeAttributeName()); //调用子类获取import配置类 String[] imports = selectImports(adviceMode); if (imports == null) { throw new IllegalArgumentException(String.format("Unknown AdviceMode: '%s'", adviceMode)); } return imports; } /** * 根据mode值返回import类完全限定名的数组 */ protected abstract String[] selectImports(AdviceMode adviceMode);}看完AsyncConfigurationSelector类的结构，再看下这个类是在哪实例化的，然后再看调用了哪些方法。
AsyncConfigurationSelector是在ConfigurationClassParser的processImports方法实例化的，而获取所有Import是通过ConfigurationClassParser的Set getImports(SourceClass sourceClass)方法，从启动类注解开始，递归遍历所有注解上的@Import注解，获取@Import注解的value值。ConfigurationClassParser的processImports方法循环遍历所有Import的value值，对每一个调用上图578行代码实例化selector，实例化每个selector后，会继续调用每个selector的selectImports(adviceMode)方法，获取到每个selector配置的imports配置类，对这些配置类继续递归调用ConfigurationClassParser的processImports方法实例化每个import配置类。这样就实现了@Import注解的功能，根据配置导入相应的配置类。
关于@Import可以参考这篇文章 @Import注解使用
实例化ProxyAsyncConfiguration
@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(); Class<? extends Annotation> customAsyncAnnotation = this.enableAsync.getClass("annotation"); if (customAsyncAnnotation != AnnotationUtils.getDefaultValue(EnableAsync.class, "annotation")) { bpp.setAsyncAnnotationType(customAsyncAnnotation); } if (this.executor != null) { bpp.setExecutor(this.executor); } if (this.exceptionHandler != null) { bpp.setExceptionHandler(this.exceptionHandler); } bpp.setProxyTargetClass(this.enableAsync.getBoolean("proxyTargetClass")); bpp.setOrder(this.enableAsync.<Integer>getNumber("order")); return bpp; }}ProxyAsyncConfiguration这个类主要功能是声明了AsyncAnnotationBeanPostProcessor这个Bean，就是异步注解后置处理器，首先需要看下AsyncAnnotationBeanPostProcessor这个类的结构
Serializable序列化接口，ProxyConfig代理相关的统一配置类，Ordered顺序优先级接口，AopInfrastructureBean是标识当前这个类是基础的类不允许被代理。
Aware接口是应用程序感知Spring框架一些功能的接口。比如BeanFactoryAware接口能获取Spring的BeanFactory，实现了该接口的类在实例化时，Spring框架会将BeanFactory传到重写的setBeanFactory(BeanFactory beanFactory)方法里面。BeanClassLoaderAware接口能够获取Spring的BeanClassLoader。关于Aware可以参考这篇文章 Spring中的aware接口
BeanPostProcessor接口是Bean的后置处理器，在Bean实例化后、属性设置完毕，自定义初始化方法执行之前和之后进行bean的处理。关于Spring后置处理器可以看这篇文章 后置处理器的使用
ProxyProcessorSupport类主要有两个功能，第一个是实现了BeanClassLoaderAware接口设置类加载器，第二个是protected void evaluateProxyInterfaces(Class<?> beanClass, ProxyFactory proxyFactory)方法，用来判断bean是否使用基于类的代理，如果不是，把需要代理的接口加到代理工厂里面。
下面重点讲解下面三个类
AbstractAdvisingBeanPostProcessor
/** * */public abstract class AbstractAdvisingBeanPostProcessor extends ProxyProcessorSupport implements BeanPostProcessor { /** * Advisor是一个基本接口，持有一个Advice（在连接点采取的操作）和一个确定Advice使用范围的过滤器（比如切入点），是为了支持不同类型Advice提供的抽象 * 实例化AsyncAnnotationBeanPostProcessor，初始化该属性的时候，传进来的实际类型是AsyncAnnotationAdvisor */ @Nullable protected Advisor advisor; /** * 如果当前传进来的bean是Advised类型，且持有很多Advisor，是否把当前Advisor放在最前面 */ protected boolean beforeExistingAdvisors = false; /** * bean对象和是否能被当前后置处理器的advisor处理的一个Map缓存，因为同一bean会走项目所有的后置处理器，当前这个抽象类有很多子类，使用缓存可以加快效率 */ private final Map<Class<?>, Boolean> eligibleBeans = new ConcurrentHashMap<>(256); public void setBeforeExistingAdvisors(boolean beforeExistingAdvisors) { this.beforeExistingAdvisors = beforeExistingAdvisors; } //重写的初始化之前的操作，直接返回 @Override public Object postProcessBeforeInitialization(Object bean, String beanName) { return bean; } @Override public Object postProcessAfterInitialization(Object bean, String beanName) { if (this.advisor == null || bean instanceof AopInfrastructureBean) { // Ignore AOP infrastructure such as scoped proxies. return bean; } //如果当前传进来的Bean是Advised类型，把当前的advisor放到一起管理 if (bean instanceof Advised) { Advised advised = (Advised) bean; if (!advised.isFrozen() && isEligible(AopUtils.getTargetClass(bean))) { //当前Advisor到Advised中， 根据beforeExistingAdvisors是否添加到前面 if (this.beforeExistingAdvisors) { advised.addAdvisor(0, this.advisor); } else { advised.addAdvisor(this.advisor); } return bean; } } //当前传进来的Bean对象，是否能够被当前后置处理器的Advisor处理 if (isEligible(bean, beanName)) { //为传进来的bean生成一个代理工厂 ProxyFactory proxyFactory = prepareProxyFactory(bean, beanName); //如果是接口代理，调用父类ProxyProcessorSupport的方法把需要代理接口加入到代理工厂对象中 if (!proxyFactory.isProxyTargetClass()) { evaluateProxyInterfaces(bean.getClass(), proxyFactory); } //将当前advisor加入到代理工厂中，ProxyFactory也是实现了Advised接口的 proxyFactory.addAdvisor(this.advisor); //子类自定义处理代理工厂 customizeProxyFactory(proxyFactory); //生成代理对象 return proxyFactory.getProxy(getProxyClassLoader()); } // No proxy needed. return bean; } protected boolean isEligible(Object bean, String beanName) { return isEligible(bean.getClass()); } protected boolean isEligible(Class<?> targetClass) { Boolean eligible = this.eligibleBeans.get(targetClass); if (eligible != null) { return eligible; } if (this.advisor == null) { return false; } //根据不同类型的advisor中的ClassFilter和MethodMatcher，判断是否能处理当前bean的Class对象 eligible = AopUtils.canApply(this.advisor, targetClass); //放入缓存 this.eligibleBeans.put(targetClass, eligible); return eligible; } protected ProxyFactory prepareProxyFactory(Object bean, String beanName) { ProxyFactory proxyFactory = new ProxyFactory(); proxyFactory.copyFrom(this); proxyFactory.setTarget(bean); return proxyFactory; } protected void customizeProxyFactory(ProxyFactory proxyFactory) { }}有关Advisor、Advice、Advised可以看这三篇文章 Spring AOP(三) Advisor类架构 Advice、Advisor、Advised都是什么接口？ Spring AOP之Advisor、PointcutAdvisor、IntroductionAdvisor、IntroductionInterceptor
AbstractBeanFactoryAwareAdvisingPostProcessor
这个类只是多实现了BeanFactoryAware接口，为了获取BeanFactory，重写了父类的prepareProxyFactory和isEligible方法
public abstract class AbstractBeanFactoryAwareAdvisingPostProcessor extends AbstractAdvisingBeanPostProcessor implements BeanFactoryAware { /** * 除了有ConfigurableBeanFactory的功能，为分析、修改bean的定义，或单例bean预初始化提供了便利 */ @Nullable private ConfigurableListableBeanFactory beanFactory; /** * BeanFactoryAware的功能，Spring框架会将BeanFactory传进来，调用这个方法 */ @Override public void setBeanFactory(BeanFactory beanFactory) { //如果是指定的类型就初始化beanFactory这个属性 this.beanFactory = (beanFactory instanceof ConfigurableListableBeanFactory ? (ConfigurableListableBeanFactory) beanFactory : null); } @Override protected ProxyFactory prepareProxyFactory(Object bean, String beanName) { if (this.beanFactory != null) { //在BeanDefinition中设置bean的originalTargetClass属性 AutoProxyUtils.exposeTargetClass(this.beanFactory, beanName, bean.getClass()); } //调用父类方法生成代理工厂 ProxyFactory proxyFactory = super.prepareProxyFactory(bean, beanName); //即使代理工厂是基于接口代理，但是如果给定的bean和BeanFactory需要基于类代理的，设置代理工厂为基于类代理 if (!proxyFactory.isProxyTargetClass() && this.beanFactory != null && AutoProxyUtils.shouldProxyTargetClass(this.beanFactory, beanName)) { proxyFactory.setProxyTargetClass(true); } return proxyFactory; } /** * 如果beanName表示是原始实例，那么跳过任何代理，否则调用父类isEligible方法 */ @Override protected boolean isEligible(Object bean, String beanName) { return (!AutoProxyUtils.isOriginalInstance(beanName, bean.getClass()) && super.isEligible(bean, beanName)); }}AsyncAnnotationBeanPostProcessor
public class AsyncAnnotationBeanPostProcessor extends AbstractBeanFactoryAwareAdvisingPostProcessor { /** * 默认的TaskExecutor bean名称，值是"taskExecutor" */ public static final String DEFAULT_TASK_EXECUTOR_BEAN_NAME = AnnotationAsyncExecutionInterceptor.DEFAULT_TASK_EXECUTOR_BEAN_NAME; protected final Log logger = LogFactory.getLog(getClass()); //线程池的一个Supplier接口，Supplier是一个函数式接口，通过他的Get方法可以获得Executor的实例 @Nullable private Supplier<Executor> executor; /** * 处理异步方法引发的未捕获异常的策略。异步方法通常返回一个Future实例，该实例允许访问底层异常。 * 当该方法不提供该返回类型时，可以使用该处理程序来管理此类未捕获的异常。 */ @Nullable private Supplier<AsyncUncaughtExceptionHandler> exceptionHandler; /** * 自定义的异步注解，@EnableAsync的annotation属性传进来的 */ @Nullable private Class<? extends Annotation> asyncAnnotationType; //设置当前后置处理器的advisor是否在其他advisor之前 public AsyncAnnotationBeanPostProcessor() { setBeforeExistingAdvisors(true); } public void configure( @Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) { this.executor = executor; this.exceptionHandler = exceptionHandler; } public void setExecutor(Executor executor) { this.executor = SingletonSupplier.of(executor); } public void setExceptionHandler(AsyncUncaughtExceptionHandler exceptionHandler) { this.exceptionHandler = SingletonSupplier.of(exceptionHandler); } public void setAsyncAnnotationType(Class<? extends Annotation> asyncAnnotationType) { Assert.notNull(asyncAnnotationType, "'asyncAnnotationType' must not be null"); this.asyncAnnotationType = asyncAnnotationType; } @Override public void setBeanFactory(BeanFactory beanFactory) { //调用父类方法，设置ConfigurableListableBeanFactory类型的BeanFactory super.setBeanFactory(beanFactory); //生成一个Advisor实例 AsyncAnnotationAdvisor advisor = new AsyncAnnotationAdvisor(this.executor, this.exceptionHandler); if (this.asyncAnnotationType != null) { //如果有自定义的异步注解，使用自定义的异步注解 advisor.setAsyncAnnotationType(this.asyncAnnotationType); } advisor.setBeanFactory(beanFactory); //初始化advisor属性 this.advisor = advisor; }}生成AsyncAnnotationAdvisor实例需要继续往里面看，是非常关键的一步，里面有如何拦截异步方法，如何调用的逻辑，继续往下看。
AsyncAnnotationAdvisor
这个类的继承关系如下
PointcutAdvisor接口是切入点驱动的Advisor，所以AsyncAnnotationAdvisor就是持有Advice和Pointcut的Advisor，类代码如下
public class AsyncAnnotationAdvisor extends AbstractPointcutAdvisor implements BeanFactoryAware { private Advice advice; private Pointcut pointcut; public AsyncAnnotationAdvisor() { this((Supplier<Executor>) null, (Supplier<AsyncUncaughtExceptionHandler>) null); } @SuppressWarnings("unchecked") public AsyncAnnotationAdvisor( @Nullable Executor executor, @Nullable AsyncUncaughtExceptionHandler exceptionHandler) { this(SingletonSupplier.ofNullable(executor), SingletonSupplier.ofNullable(exceptionHandler)); } @SuppressWarnings("unchecked") public AsyncAnnotationAdvisor( @Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) { Set<Class<? extends Annotation>> asyncAnnotationTypes = new LinkedHashSet<>(2); //添加@Async注解 asyncAnnotationTypes.add(Async.class); try { //添加@Asynchronous注解 asyncAnnotationTypes.add((Class<? extends Annotation>) ClassUtils.forName("javax.ejb.Asynchronous", AsyncAnnotationAdvisor.class.getClassLoader())); } catch (ClassNotFoundException ex) { // If EJB 3.1 API not present, simply ignore. } //初始化advice属性 this.advice = buildAdvice(executor, exceptionHandler); //初始化pointcut属性 this.pointcut = buildPointcut(asyncAnnotationTypes); } /** * 自定义异步注解的切点，通过@EnableAsync的annotation属性传进来，默认的@Async和@javax.ejb.Asynchronous会失效 */ public void setAsyncAnnotationType(Class<? extends Annotation> asyncAnnotationType) { Assert.notNull(asyncAnnotationType, "'asyncAnnotationType' must not be null"); Set<Class<? extends Annotation>> asyncAnnotationTypes = new HashSet<>(); asyncAnnotationTypes.add(asyncAnnotationType); this.pointcut = buildPointcut(asyncAnnotationTypes); } /** * 设置advice的BeanFactory，在后面调用异步方法时，通过这个beanFactory根据executor的beanName来获取executor实例 */ @Override public void setBeanFactory(BeanFactory beanFactory) { if (this.advice instanceof BeanFactoryAware) { ((BeanFactoryAware) this.advice).setBeanFactory(beanFactory); } } @Override public Advice getAdvice() { return this.advice; } @Override public Pointcut getPointcut() { return this.pointcut; } protected Advice buildAdvice( @Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) { //生成一个advice实例，这个下面继续往里面看 AnnotationAsyncExecutionInterceptor interceptor = new AnnotationAsyncExecutionInterceptor(null); interceptor.configure(executor, exceptionHandler); return interceptor; } protected Pointcut buildPointcut(Set<Class<? extends Annotation>> asyncAnnotationTypes) { ComposablePointcut result = null; //传进来的是@Async或@javax.ejb.Asynchronous注解；也可能是自定义的一个异步注解，遍历每一个注解 for (Class<? extends Annotation> asyncAnnotationType : asyncAnnotationTypes) { //类级别的切点 Pointcut cpc = new AnnotationMatchingPointcut(asyncAnnotationType, true); //方法级别的切点 Pointcut mpc = new AnnotationMatchingPointcut(null, asyncAnnotationType, true); if (result == null) { result = new ComposablePointcut(cpc); } else { result.union(cpc); } //合并所有类级别、方法级别的切点 result = result.union(mpc); } //如果为空，返回所有方法、类都匹配的切点 return (result != null ? result : Pointcut.TRUE); }}下面需要讲下AnnotationAsyncExecutionInterceptor和AnnotationMatchingPointcut两个类，看完这两个类，整个异步处理就基本结束了
AnnotationAsyncExecutionInterceptor
AnnotationAsyncExecutionInterceptor类文件结构如下
Interceptor是通用拦截器接口抽象，是个空接口；
MethodInterceptor是方法拦截器，有一个Object invoke(MethodInvocation invocation)方法，重写该方法可以在方法调用的前后，加入自定义逻辑，下面是源码中提供的例子 （关于方法拦截器可以参考这篇文章 Spring方法拦截器MethodInterceptor）
class TracingInterceptor implements MethodInterceptor { Object invoke(MethodInvocation i) throws Throwable { System.out.println("method "+i.getMethod()+" is called on "+ i.getThis()+" with args "+i.getArguments()); Object ret=i.proceed();//调用目标方法 System.out.println("method "+i.getMethod()+" returns "+ret); return ret; }}AsyncExecutionAspectSupport是异步方法执行切面的基类，实现了BeanFactoryWare接口，可以获取Bean工厂，源码如下
public abstract class AsyncExecutionAspectSupport implements BeanFactoryAware { //线程池默认bean名称 public static final String DEFAULT_TASK_EXECUTOR_BEAN_NAME = "taskExecutor"; //CompletableFuture类是否存在，这个类是java8引入的，这个字段doSubmit方法有用到 private static final boolean completableFuturePresent = ClassUtils.isPresent( "java.util.concurrent.CompletableFuture", AsyncExecutionInterceptor.class.getClassLoader()); protected final Log logger = LogFactory.getLog(getClass()); //异步方法和对应线程池实例的缓存，因为每个异步方法可以指定线程池实例 private final Map<Method, AsyncTaskExecutor> executors = new ConcurrentHashMap<Method, AsyncTaskExecutor>(16); //默认的线程池实例 private volatile Executor defaultExecutor; //未捕获异常的处理器 private AsyncUncaughtExceptionHandler exceptionHandler; private BeanFactory beanFactory; public AsyncExecutionAspectSupport(Executor defaultExecutor) { this(defaultExecutor, new SimpleAsyncUncaughtExceptionHandler()); } public AsyncExecutionAspectSupport(Executor defaultExecutor, AsyncUncaughtExceptionHandler exceptionHandler) { this.defaultExecutor = defaultExecutor; this.exceptionHandler = exceptionHandler; } public void setExecutor(Executor defaultExecutor) { this.defaultExecutor = defaultExecutor; } public void setExceptionHandler(AsyncUncaughtExceptionHandler exceptionHandler) { this.exceptionHandler = exceptionHandler; } /** * 重写BeanFactoryAware接口的方法，设置Bean工厂 */ @Override public void setBeanFactory(BeanFactory beanFactory) { this.beanFactory = beanFactory; } /** * 根据指定异步方法获取对应的线程池实例 */ protected AsyncTaskExecutor determineAsyncExecutor(Method method) { //从缓存里面获取，如果获取得到直接返回 AsyncTaskExecutor executor = this.executors.get(method); if (executor == null) { Executor targetExecutor; //根据方法获取线程池实例的Bean名称，@Async的value属性的值 String qualifier = getExecutorQualifier(method); if (StringUtils.hasLength(qualifier)) { //bean工厂根据bean名称获取线程池实例 targetExecutor = findQualifiedExecutor(this.beanFactory, qualifier); } else { targetExecutor = this.defaultExecutor; if (targetExecutor == null) { synchronized (this.executors) { if (this.defaultExecutor == null) { this.defaultExecutor = getDefaultExecutor(this.beanFactory); } targetExecutor = this.defaultExecutor; } } } if (targetExecutor == null) { return null; } //如果不是AsyncListenableTaskExecutor类型的线程池实例，构造一个TaskExecutorAdapter实例，TaskExecutorAdapter是带一个TaskDecorator属性的线程池实例，可以对要执行的任务进行装饰，比如SpringSecurity进行权限管理时，创建异步任务会丢失父线程的权限信息，可以写一个类实现TaskDecorator接口，在decorate方法里面往SecurityContextHolder设置上下文信息 executor = (targetExecutor instanceof AsyncListenableTaskExecutor ? (AsyncListenableTaskExecutor) targetExecutor : new TaskExecutorAdapter(targetExecutor)); this.executors.put(method, executor); } return executor; } /** * 抽象方法，根据异步方法获取线程池bean的名称 */ protected abstract String getExecutorQualifier(Method method); /** * 根据bean名称获取Executor类型的线程池实例 */ protected Executor findQualifiedExecutor(BeanFactory beanFactory, String qualifier) { if (beanFactory == null) { throw new IllegalStateException("BeanFactory must be set on " + getClass().getSimpleName() + " to access qualified executor '" + qualifier + "'"); } return BeanFactoryAnnotationUtils.qualifiedBeanOfType(beanFactory, Executor.class, qualifier); } //获取默认的线程池实例 protected Executor getDefaultExecutor(BeanFactory beanFactory) { if (beanFactory != null) { try { // 找TaskExecutor类型的线程池实例 return beanFactory.getBean(TaskExecutor.class); } catch (NoUniqueBeanDefinitionException ex) { logger.debug("Could not find unique TaskExecutor bean", ex); try { //找名称为taskExecutor的线程池实例 return beanFactory.getBean(DEFAULT_TASK_EXECUTOR_BEAN_NAME, Executor.class); } catch (NoSuchBeanDefinitionException ex2) { if (logger.isInfoEnabled()) { logger.info("More than one TaskExecutor bean found within the context, and none is named " + "'taskExecutor'. Mark one of them as primary or name it 'taskExecutor' (possibly " + "as an alias) in order to use it for async processing: " + ex.getBeanNamesFound()); } } } catch (NoSuchBeanDefinitionException ex) { logger.debug("Could not find default TaskExecutor bean", ex); try { //找名称为taskExecutor的线程池实例 return beanFactory.getBean(DEFAULT_TASK_EXECUTOR_BEAN_NAME, Executor.class); } catch (NoSuchBeanDefinitionException ex2) { logger.info("No task executor bean found for async processing: " + "no bean of type TaskExecutor and no bean named 'taskExecutor' either"); } // Giving up -> either using local default executor or none at all... } } return null; } /** * 执行异步任务，参数分别是异步方法执行逻辑、线程池实例、异步方法返回结果 */ protected Object doSubmit(Callable<Object> task, AsyncTaskExecutor executor, Class<?> returnType) { //如果是Java8，使用CompletableFuture来执行异步任务 if (completableFuturePresent) { Future<Object> result = CompletableFutureDelegate.processCompletableFuture(returnType, task, executor); if (result != null) { return result; } } //如果返回类型是ListenableFuture调用submitListenable if (ListenableFuture.class.isAssignableFrom(returnType)) { return ((AsyncListenableTaskExecutor) executor).submitListenable(task); } //如果返回类型是其他的Future类型，直接交给线程池执行 else if (Future.class.isAssignableFrom(returnType)) { return executor.submit(task); } else { //直接不返回结果 executor.submit(task); return null; } } //异常处理 protected void handleError(Throwable ex, Method method, Object... params) throws Exception { //带返回值的直接抛出异常 if (Future.class.isAssignableFrom(method.getReturnType())) { ReflectionUtils.rethrowException(ex); } else { //异常处理器处理异常，即使再出现异常也不抛出 try { this.exceptionHandler.handleUncaughtException(ex, method, params); } catch (Throwable ex2) { logger.error("Exception handler for async method '" + method.toGenericString() + "' threw unexpected exception itself", ex2); } } } /** * Java8下执行异步任务的内部类 */ @UsesJava8 private static class CompletableFutureDelegate { public static <T> Future<T> processCompletableFuture(Class<?> returnType, final Callable<T> task, Executor executor) { //如果异步方法返回值不是CompletableFuture类型直接返回null if (!CompletableFuture.class.isAssignableFrom(returnType)) { return null; } //调用CompletableFuture的supplyAsync方法去执行task任务 return CompletableFuture.supplyAsync(new Supplier<T>() { @Override public T get() { try { return task.call(); } catch (Throwable ex) { throw new CompletionException(ex); } } }, executor); } }}关于CompletableFuture类可以参考这篇文章 CompletableFuture原理解析
下面看AsyncExecutionInterceptor这个类，这个类是处理异步方法调用的方法拦截器
public class AsyncExecutionInterceptor extends AsyncExecutionAspectSupport implements MethodInterceptor, Ordered { public AsyncExecutionInterceptor(Executor defaultExecutor) { super(defaultExecutor); } public AsyncExecutionInterceptor(Executor defaultExecutor, AsyncUncaughtExceptionHandler exceptionHandler) { super(defaultExecutor, exceptionHandler); } /** * MethodInterceptor重写的方法，方法调用前后处理一些逻辑 */ @Override public Object invoke(final MethodInvocation invocation) throws Throwable { //获取invocation的目标对象的class对象（被调用的异步方法所属对象的Class对象） Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null); //通过class对象和invocation的method获取Method Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass); //没理解为什么还要再获取一次，我debug时候调用specificMethod.equlas(userDeclaredMethod)返回的是true final Method userDeclaredMethod = BridgeMethodResolver.findBridgedMethod(specificMethod); //通过method获取处理这个异步方法的线程池实例 AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod); if (executor == null) { throw new IllegalStateException( "No executor specified and no default executor set on AsyncExecutionInterceptor either"); } //将异步方法封装成一个Callable对象 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; } }; //把异步任务、线程池实例、返回值类型传进去，调用父类的AsyncExecutionAspectSupport的doSubmit方法 return doSubmit(task, executor, invocation.getMethod().getReturnType()); } /** * 根据异步方法，获取处理该异步方法的线程池实例的bean名称，后续在BeanFactory里面根据这个名称获取线程池实例，本类中返回null，子类会重写这个方法，AnnotationAsyncExecutionInterceptor重写改方法是获取@Async注解的value值 */ @Override protected String getExecutorQualifier(Method method) { return null; } /** * 调用父类的获取默认线程池实例的方法，如果获取不到，使用SimpleAsyncTaskExecutor实例 * SimpleAsyncTaskExecutor这个线程池会为每个任务触发一个新线程，异步执行它，相当于没用线程池 */ @Override protected Executor getDefaultExecutor(BeanFactory beanFactory) { Executor defaultExecutor = super.getDefaultExecutor(beanFactory); return (defaultExecutor != null ? defaultExecutor : new SimpleAsyncTaskExecutor()); } @Override public int getOrder() { return Ordered.HIGHEST_PRECEDENCE; }}最后看最终实例化的AnnotationAsyncExecutionInterceptor这个类
public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> { private static final String ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME = "org.springframework.scheduling.aspectj.AspectJAsyncConfiguration"; /** * 根据@EnableAsync的mode属性返回不同配置类 */ @Override public String[] selectImports(AdviceMode adviceMode) { switch (adviceMode) { case PROXY: return new String[] { ProxyAsyncConfiguration.class.getName() }; case ASPECTJ: return new String[] { ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME }; default: return null; } }}/** * 基于注解的mode属性来获取imports的基类 */public abstract class AdviceModeImportSelector<A extends Annotation> implements ImportSelector { public static final String DEFAULT_ADVICE_MODE_ATTRIBUTE_NAME = "mode"; protected String getAdviceModeAttributeName() { return DEFAULT_ADVICE_MODE_ATTRIBUTE_NAME; } /** * 获取泛型注解的mode属性，调用子类的selectImports(AdviceMode adviceMode)方法获取import配置类 * AnnotationMetadata importingClassMetadata是SpringBoot启动类的上获取的注解（我当前项目上就@EnableAsync, @SpringBootApplication两个注解）, ConfigurationClassParser的processImports方法传进来的 */ @Override public final String[] selectImports(AnnotationMetadata importingClassMetadata) { //获取当前类的泛型参数（我自己Debug时就是@EnableAsync, getClass()获取到的是AsyncConfigurationSelector的class对象） Class<?> annoType = GenericTypeResolver.resolveTypeArgument(getClass(), AdviceModeImportSelector.class); //获取指定当前泛型注解属性和值 AnnotationAttributes attributes = AnnotationConfigUtils.attributesFor(importingClassMetadata, annoType); if (attributes == null) { throw new IllegalArgumentException(String.format( "@%s is not present on importing class '%s' as expected", annoType.getSimpleName(), importingClassMetadata.getClassName())); } //获取mode属性值 AdviceMode adviceMode = attributes.getEnum(this.getAdviceModeAttributeName()); //调用子类获取import配置类 String[] imports = selectImports(adviceMode); if (imports == null) { throw new IllegalArgumentException(String.format("Unknown AdviceMode: '%s'", adviceMode)); } return imports; } /** * 根据mode值返回import类完全限定名的数组 */ protected abstract String[] selectImports(AdviceMode adviceMode);}0到这里AsyncAnnotationAdvisor的advice属性已经讲完了，下面再看另外一个重要属性pointcut
AnnotationMatchingPointcut
在AsyncAnnotationAdvisor类中的Pointcut buildPointcut(Set<Class<? extends Annotation>> asyncAnnotationTypes)这个方法，我们传进来了@Async和@javax.ejb.Asynchronous注解，对每个注解，在类和方法级别都创建了AnnotationMatchingPointcut对象，并通过ComposablePointcut对象把所有的切点组合在一起，AnnotationMatchingPointcut源码如下（关于ClassFilter可以参考这篇文章 spring-aop 组件详解 ——ClassFilter 类过滤器 ）
public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> { private static final String ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME = "org.springframework.scheduling.aspectj.AspectJAsyncConfiguration"; /** * 根据@EnableAsync的mode属性返回不同配置类 */ @Override public String[] selectImports(AdviceMode adviceMode) { switch (adviceMode) { case PROXY: return new String[] { ProxyAsyncConfiguration.class.getName() }; case ASPECTJ: return new String[] { ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME }; default: return null; } }}/** * 基于注解的mode属性来获取imports的基类 */public abstract class AdviceModeImportSelector<A extends Annotation> implements ImportSelector { public static final String DEFAULT_ADVICE_MODE_ATTRIBUTE_NAME = "mode"; protected String getAdviceModeAttributeName() { return DEFAULT_ADVICE_MODE_ATTRIBUTE_NAME; } /** * 获取泛型注解的mode属性，调用子类的selectImports(AdviceMode adviceMode)方法获取import配置类 * AnnotationMetadata importingClassMetadata是SpringBoot启动类的上获取的注解（我当前项目上就@EnableAsync, @SpringBootApplication两个注解）, ConfigurationClassParser的processImports方法传进来的 */ @Override public final String[] selectImports(AnnotationMetadata importingClassMetadata) { //获取当前类的泛型参数（我自己Debug时就是@EnableAsync, getClass()获取到的是AsyncConfigurationSelector的class对象） Class<?> annoType = GenericTypeResolver.resolveTypeArgument(getClass(), AdviceModeImportSelector.class); //获取指定当前泛型注解属性和值 AnnotationAttributes attributes = AnnotationConfigUtils.attributesFor(importingClassMetadata, annoType); if (attributes == null) { throw new IllegalArgumentException(String.format( "@%s is not present on importing class '%s' as expected", annoType.getSimpleName(), importingClassMetadata.getClassName())); } //获取mode属性值 AdviceMode adviceMode = attributes.getEnum(this.getAdviceModeAttributeName()); //调用子类获取import配置类 String[] imports = selectImports(adviceMode); if (imports == null) { throw new IllegalArgumentException(String.format("Unknown AdviceMode: '%s'", adviceMode)); } return imports; } /** * 根据mode值返回import类完全限定名的数组 */ protected abstract String[] selectImports(AdviceMode adviceMode);}1异步方法调用阶段
方法调用时，会进入DynamicAdvisedInterceptor这个类的Object intercept(Object proxy, Method method, Object[] args, MethodProxy methodProxy)方法，为什么会进这个方法，还需要往前看，在应用启动阶段生成代理对象时做的具体操作。在AbstractAdvisingBeanPostProcessor类中，
红框处生成代理对象时，会再调用CglibAopProxy类的getProxy方法
里面会再调用getCallbacks方法获取回调
在获取回调的方法中会直接new一个DynamicAdvisedInterceptor，把这个方法拦截器设置到回调中，在调用异步方法时会进入这个拦截器，DynamicAdvisedInterceptor的源码如下
public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> { private static final String ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME = "org.springframework.scheduling.aspectj.AspectJAsyncConfiguration"; /** * 根据@EnableAsync的mode属性返回不同配置类 */ @Override public String[] selectImports(AdviceMode adviceMode) { switch (adviceMode) { case PROXY: return new String[] { ProxyAsyncConfiguration.class.getName() }; case ASPECTJ: return new String[] { ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME }; default: return null; } }}/** * 基于注解的mode属性来获取imports的基类 */public abstract class AdviceModeImportSelector<A extends Annotation> implements ImportSelector { public static final String DEFAULT_ADVICE_MODE_ATTRIBUTE_NAME = "mode"; protected String getAdviceModeAttributeName() { return DEFAULT_ADVICE_MODE_ATTRIBUTE_NAME; } /** * 获取泛型注解的mode属性，调用子类的selectImports(AdviceMode adviceMode)方法获取import配置类 * AnnotationMetadata importingClassMetadata是SpringBoot启动类的上获取的注解（我当前项目上就@EnableAsync, @SpringBootApplication两个注解）, ConfigurationClassParser的processImports方法传进来的 */ @Override public final String[] selectImports(AnnotationMetadata importingClassMetadata) { //获取当前类的泛型参数（我自己Debug时就是@EnableAsync, getClass()获取到的是AsyncConfigurationSelector的class对象） Class<?> annoType = GenericTypeResolver.resolveTypeArgument(getClass(), AdviceModeImportSelector.class); //获取指定当前泛型注解属性和值 AnnotationAttributes attributes = AnnotationConfigUtils.attributesFor(importingClassMetadata, annoType); if (attributes == null) { throw new IllegalArgumentException(String.format( "@%s is not present on importing class '%s' as expected", annoType.getSimpleName(), importingClassMetadata.getClassName())); } //获取mode属性值 AdviceMode adviceMode = attributes.getEnum(this.getAdviceModeAttributeName()); //调用子类获取import配置类 String[] imports = selectImports(adviceMode); if (imports == null) { throw new IllegalArgumentException(String.format("Unknown AdviceMode: '%s'", adviceMode)); } return imports; } /** * 根据mode值返回import类完全限定名的数组 */ protected abstract String[] selectImports(AdviceMode adviceMode);}2DynamicAdvisedInterceptor这个类有一个AdvisedSupport属性，这个属性其实是从异步注解后置处理器的advisor拿过来，然后执行new CglibMethodInvocation那一行代码，把之前设置在后置处理器中的Advice传进去，接着会跳到AsyncExecutionInterceptor的invoke方法去，把任务放到线程池汇中去执行，然后处理返回值，整个流程就结束了。
总结与启示
整个流程就是，首先在项目启动阶段，通过@EnableAsync注解导入一些配置类，最终实例化一个异步注解的后置处理器（AsyncAnnotationBeanPostProcessor），这个后置处理器有自己的切入点和处理逻辑，会拦截项目中所有的bean，如果某个bean符合该后置处理器的切入点，那么SpringBoot会通过AOP生成一个代理对象，生成代理对象时会设置一个回调，回调的内容就是后置处理器中的处理逻辑（实际逻辑就是将异步方法内容放入线程池中执行），并将这个代理对象注入到使用的地方。
当真正调用异步方法时，因为注入的是代理对象，那么调用到异步方法之前会进入之前设置的回调，去执行异步方法内容，执行完毕后会根据不同的返回值类型处理返回值，至此异步方法就执行完毕了。整篇文章还是有点抽象，过程描述不太详细，只是描述每个类的功能，之前想画一下时序图，但是因为方法调用太多画的图会很复杂，所以强烈建议读者在应用启动阶段和方法调用阶段设置断点，一步步Debug会理解得更深刻。
看完@Async源码感触还是挺多的，首先就是源码的类结构的设计真的很优秀，很多思路可以借鉴，引入到自己的项目中。其次看源码我的一个思路就是，从上往下，从父类到子类，父接口到子接口，对于每一个类，确定是如何实例化的，每个属性是如何设置的，有哪些方法，每个方法的每一行是做什么的，最后大致看完以后，设置断点一步步Debug走下去，边Debug边看每个类的结构，这样对某个类就有个整体把握。
当然源码很复杂，有些地方不可能看的很细，在我看@Async源码的时候有些地方也没有理解，比如AOP那一块，以后有机会再慢慢看看，总之看源码慢慢来，看完有收获就可以。
原文：https://juejin.cn/post/7097521638119309319