Spring @Async异步任务源码(1)—＜task:annotation-driven/＞、＜task:executor/＞、@EnableAsync解析

た入场券 2022-10-31 12:18 432阅读 0赞

>   基于最新Spring 5.x，详细介绍了Spring @Async异步任务机制的实现原理！主要是< task:annotation-driven/>、< task:executor/>、@EnableAsync解析。

此前我们学习了Spring异步任务的概念和用法：[Spring 5.x 学习(7)—@Async异步任务机制应用详解][Spring 5.x _7_Async]，现在我们来深入学习它的源码！  
  在此前学习的时候，我们通过`<task:annotation-driven/>`标签或者`@EnableAsync`注解来开启异步任务的支持，我们能够想到这个标签或者注解就是Spring异步任务的解析入口，因此我们这两方面入手来学习源码！

### 文章目录 ###

*  1 < task:annotation-driven/>标签解析
 *   *  1.1 < task:annotation-driven/>标签简介
     *  1.2 < task:annotation-driven/>标签源码
 *  2 < task:executor/>标签解析
 *   *  2.1 getBeanClassName获取bean定义的class类型
     *  2.2 doParse解析< task:executor/>标签属性
     *  2.3 TaskExecutorFactoryBean任务执行器工厂
     *   *  2.3.1 executor.afterPropertiesSet执行器初始化
 *  3 @EnableAsync注解解析
 *   *  3.1 @EnableAsync注解简介
     *  3.2 @EnableAsync注解源码
     *   *  3.2.1 ProxyAsyncConfiguration
 *  4 小结

# 1 < task:annotation-driven/>标签解析 #

## 1.1 < task:annotation-driven/>标签简介 ##

**该标签用于启用检测任何 Spring管理的bean上的@Async和@Scheduled注解（@Async表示异步任务，@Scheduled表示定时任务，也就是说该标签有两个功能）。如果存在任何一个注解，那么将生成一个代理对象以异步的方式执行方法。Spring 3.1开始可使用@EnableAsync和@EnableScheduling注解替代该标签。**  
  此前的文章中我们已经学习过这个标签了，这里仅仅简单的复习一下该标签的属性！  
  `exception-handler`：指定一个org.springframe.aop.interceptor.AsyncUncaughtExceptionHandler 类型的异常处理器实例，用于处理在异步方法执行期间引发的异常（可以获取抛出的异常、抛出异常的异步任务方法、异步任务方法的参数）。

/**
     * @author lx
     */
    @Component
    public class MyAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
        
    
        @Override
        public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        
            System.out.println("调用异步任务出错了, message : " + ex.getMessage());
            System.out.println("方法参数, params: " + Arrays.toString(params));
            System.out.println("方法, method: " + method);
        }
    }

如果没有指定异常处理器，那么默认采用一个org.springframe.aop.interceptor.SimpleAsyncUncaughtExceptionHandehandler 的异常处理器实例，该异常处理器将会捕获异常并且输出异常日志（并非抛出异常）：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center]  
  `executor`：指定在调用异步方法时要使用的 java.util.Executor 执行器实例。 如果未提供，则默认情况下将使用org.springframework.core.task.SimpleAsyncTaskExecutor的实例。自 Spring 3.1.2 起，单个 @Async 方法可以指定要使用哪个执行器，这意味着此处指定的执行器将会充当所有 没有指定执行器的@Async 方法的默认执行器。  
  `mode`：采用Spring AOP来对异步方法进行动态代理，还是采用AspectJ来进行静态织入，默认为proxy，即Spring AOP代理，如果设置为aspectj，那么还需要spring-aspects.jar（其内部包含了aspectjweaver依赖）。  
  `proxy-target-class`：是否创建基于类的（CGLIB）代理？默认情况下，将创建基于 Java 接口的标准代理。基于类的代理要求在类上定义@Async注解。在这种情况下，接口中的注解将不起作用（它们只能与基于接口的代理一起工作）！  
  `scheduler`：指定在调用scheduled方法时使用的org.springframework.scheduling.TaskScheduler或java.util.ScheduledExecutorService实例。如果未提指定，则使用单个线程的任务计划执行器，这是一个用于调度任务的属性。

## 1.2 < task:annotation-driven/>标签源码 ##

在此前讲解Spring初始化源码：[Spring IoC容器初始化源码(3)—parseDefaultElement、parseCustomElement解析默认、扩展标签，registerBeanDefinition注册Bean定义][Spring IoC_3_parseDefaultElement_parseCustomElement_registerBeanDefinition_Bean]时就讲过，task属性属于Spring的自定义命名空间，因此它是使用`parseCustomElement`方法解析，该方法首先会找到此命名空间对应的Handler，即`TaskNamespaceHandler`，其内部保存了task命名空间下的各种标签的解析器：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center 1]  
  很容易找到< task:annotation-driven/>标签的解析器就是org.springframework.scheduling.config.AnnotationDrivenBeanDefinitionParser，最终，该标签就是通过该解析器的parse方法完成解析的！  
  parse方法非常的简单：**解析< task:annotation-driven/>标签元素的属性，并尝试注册两个bean定义：一个名为"org.springframework.context.annotation.internalAsyncAnnotationProcessor"的AsyncAnnotationBeanPostProcessor，用于异步任务的代理对象的创建；一个名为"org.springframework.context.annotation.internalScheduledAnnotationProcessor"的ScheduledAnnotationBeanPostProcessor，用于调度任务的代理对象的创建。**  
  **这两个bean定义都是BeanPostProcessor的实现，因此将会在普通bean实例化之前被实例化，并且在每个普通bean实例化之后用于判断该bean是否需要进行代理以及进行代理对象的创建工作。也就是说，这真的代理对象的创建是靠这两个后处理器来完成的，< task:annotation-driven/>标签仅仅是解析属性并注册这两个后处理器而已！**

/**
     * AnnotationDrivenBeanDefinitionParser的方法
     * <p>
     * 解析<task:annotation-driven/>标签元素，尝试注册两个bean定义
     * <p>
     * 一个名为"org.springframework.context.annotation.internalAsyncAnnotationProcessor"的AsyncAnnotationBeanPostProcessor
     * 一个名为"org.springframework.context.annotation.internalScheduledAnnotationProcessor"的ScheduledAnnotationBeanPostProcessor
     * <p>
     * 他们都是BeanPostProcessor的实现，将会在普通bean实例化之前被实例化，并且在普通bean实例化之后用于判断该bean是否需要进行代理以及进行代理对象的创建工作
     * 也就是说，这真的代理对象的创建是靠这两个后处理器来完成的，<task:annotation-driven/>标签仅仅是解析属性并注册这两个后处理器而已！
     *
     * @param element       当前<task:annotation-driven/>标签元素
     * @param parserContext 解析上下文，从中可以获取各种配置
     * @return 返回null
     */
    @Override
    @Nullable
    public BeanDefinition parse(Element element, ParserContext parserContext) {
        
        //获取源
        Object source = parserContext.extractSource(element);
    
        //为<task:annotation-driven> 标签注册一个CompositeComponentDefinition类型的bean定义组件
        //该类型的bean定义，内部可以包含多个bean定义，因此又称为复合bean定义
        CompositeComponentDefinition compDefinition = new CompositeComponentDefinition(element.getTagName(), source);
        //存入解析上下文内部的containingComponents集合中，入栈顶
        parserContext.pushContainingComponent(compDefinition);
    
        //获取上下文中保存的当前容器的注册表
        BeanDefinitionRegistry registry = parserContext.getRegistry();
        //获取mode属性
        String mode = element.getAttribute("mode");
        if ("aspectj".equals(mode)) {
        
            // mode="aspectj"
            //如果设置为aspectj，那么注册一个Aspect异步执行切面的 bean 。
            registerAsyncExecutionAspect(element, parserContext);
        } else {
        
            // mode="proxy"
            //如果设置为proxy，那么采用Spring AOP来进行代理
    
            /*
             * 注册用于Async异步任务的bean后处理器AsyncAnnotationBeanPostProcessor
             */
    
            //如果bean注册表中已经包含了名为"org.springframework.context.annotation.internalAsyncAnnotationProcessor"的bean定义
            if (registry.containsBeanDefinition(TaskManagementConfigUtils.ASYNC_ANNOTATION_PROCESSOR_BEAN_NAME)) {
        
                //那么直接抛出异常，表示可能出现了多个AsyncAnnotationBeanPostProcessor
                parserContext.getReaderContext().error(
                        "Only one AsyncAnnotationBeanPostProcessor may exist within the context.", source);
            }
            //否则，注册一个名为"org.springframework.context.annotation.internalAsyncAnnotationProcessor"的bean定义
            else {
        
                //获取bean定义构建者，设置beanClassName为"org.springframework.scheduling.annotation.AsyncAnnotationBeanPostProcessor"
                //AsyncAnnotationBeanPostProcessor实际上是一个bean后处理器
                BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
                        "org.springframework.scheduling.annotation.AsyncAnnotationBeanPostProcessor");
                builder.getRawBeanDefinition().setSource(source);
    
                /*解析标签的executor属性，并设置为要创建的bean定义的executor属性*/
                String executor = element.getAttribute("executor");
                if (StringUtils.hasText(executor)) {
        
                    builder.addPropertyReference("executor", executor);
                }
                /*解析标签的exception-handler属性，并设置为要创建的bean定义的exceptionHandler属性*/
                String exceptionHandler = element.getAttribute("exception-handler");
                if (StringUtils.hasText(exceptionHandler)) {
        
                    builder.addPropertyReference("exceptionHandler", exceptionHandler);
                }
                /*解析标签的proxy-target-class属性，并设置为要创建的bean定义的proxyTargetClass属性*/
                if (Boolean.parseBoolean(element.getAttribute(AopNamespaceUtils.PROXY_TARGET_CLASS_ATTRIBUTE))) {
        
                    builder.addPropertyValue("proxyTargetClass", true);
                }
                /*
                 * 注册bean后处理器到注册表中，名为"org.springframework.context.annotation.internalAsyncAnnotationProcessor"
                 */
                registerPostProcessor(parserContext, builder, TaskManagementConfigUtils.ASYNC_ANNOTATION_PROCESSOR_BEAN_NAME);
            }
        }
        /*
         * 注册用于Scheduled调度任务的bean后处理器ScheduledAnnotationBeanPostProcessor
         */
        //如果注册表包含名为"org.springframework.context.annotation.internalScheduledAnnotationProcessor"的bean定义
        if (registry.containsBeanDefinition(TaskManagementConfigUtils.SCHEDULED_ANNOTATION_PROCESSOR_BEAN_NAME)) {
        
            //那么直接抛出异常，表示可能出现了多个ScheduledAnnotationBeanPostProcessor
            parserContext.getReaderContext().error(
                    "Only one ScheduledAnnotationBeanPostProcessor may exist within the context.", source);
        }
        //否则，注册一个名为"org.springframework.context.annotation.internalScheduledAnnotationProcessor"的bean定义
        else {
        
            //获取bean定义构建者，设置beanClassName为"org.springframework.scheduling.annotation.ScheduledAnnotationBeanPostProcessor"
            //ScheduledAnnotationBeanPostProcessor实际上是一个bean后处理器
            BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
                    "org.springframework.scheduling.annotation.ScheduledAnnotationBeanPostProcessor");
            builder.getRawBeanDefinition().setSource(source);
    
            /*解析标签的scheduler属性，并设置为要创建的bean定义的scheduler属性*/
            String scheduler = element.getAttribute("scheduler");
            if (StringUtils.hasText(scheduler)) {
        
                builder.addPropertyReference("scheduler", scheduler);
            }
            /*
             * 注册bean后处理器到注册表中，名为"org.springframework.context.annotation.internalScheduledAnnotationProcessor"
             */
            registerPostProcessor(parserContext, builder, TaskManagementConfigUtils.SCHEDULED_ANNOTATION_PROCESSOR_BEAN_NAME);
        }
    
        //出栈并注册，并不是注册到注册表中……可能发布事件或者什么也不做
        parserContext.popAndRegisterContainingComponent();
        //返回null
        return null;
    }
    
    /**
     * AnnotationDrivenBeanDefinitionParser的方法
     * <p>
     * 尝试注册bean后处理器
     *
     * @param parserContext 解析上下文
     * @param builder       bean定义构建者
     * @param beanName      注册的beanName
     */
    private static void registerPostProcessor(
            ParserContext parserContext, BeanDefinitionBuilder builder, String beanName) {
        
        builder.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
        //调用registerBeanDefinition方法注册bean定义，该方法我们在此前就讲过了
        parserContext.getRegistry().registerBeanDefinition(beanName, builder.getBeanDefinition());
        //获取bdholder
        BeanDefinitionHolder holder = new BeanDefinitionHolder(builder.getBeanDefinition(), beanName);
        //注册组件，并不是注册到注册表中……而是发布组件注册事件，可以被监听到
        parserContext.registerComponent(new BeanComponentDefinition(holder));
    }

# 2 < task:executor/>标签解析 #

`<task:executor/>`标签用于指定一个自定义的任务执行器，该标签本身比较简单，它是由`ExecutorBeanDefinitionParser`解析器来解析的！  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center 2]  
  ExecutorBeanDefinitionParser继承了AbstractSingleBeanDefinitionParser，在此前介绍[< context:property-placeholder/>源码][context_property-placeholder]的时候就讲过此体系，主要用于解析单个bean定义的标签，标签将会被解析为一个bean定义  
  **我们主要看ExecutorBeanDefinitionParser自身重写的方法！**

## 2.1 getBeanClassName获取bean定义的class类型 ##

该方法用于获取bean定义的class类型！

@Override
    protected String getBeanClassName(Element element) {
        
       return "org.springframework.scheduling.config.TaskExecutorFactoryBean";
    }

对于< task:executor/>标签，将会被解析为class类型为TaskExecutorFactoryBean的bean定义，它是一个FactoryBean，可以从里面的getObject方法获取所需的执行器！并且bean定义的id就是< task:executor/>标签的id。

## 2.2 doParse解析< task:executor/>标签属性 ##

**该方法就是解析< task:executor/>标签的自有属性，还是很简单的。需要注意的是，这些属性是TaskExecutorFactoryBean的属性，并不是直接设置的TaskExecutor的属性。**

/**
     * 解析<task:executor/>标签的自有属性
     *
     * @param element       当前<task:executor/>标签元素
     * @param parserContext 解析上下文.,包含各种常用变量
     * @param builder       bean定义构建者
     */
    @Override
    protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
        
        //设置keepAliveSeconds属性
        String keepAliveSeconds = element.getAttribute("keep-alive");
        if (StringUtils.hasText(keepAliveSeconds)) {
        
            builder.addPropertyValue("keepAliveSeconds", keepAliveSeconds);
        }
        //设置queueCapacity属性
        String queueCapacity = element.getAttribute("queue-capacity");
        if (StringUtils.hasText(queueCapacity)) {
        
            builder.addPropertyValue("queueCapacity", queueCapacity);
        }
        //设置rejectedExecutionHandler拒绝策略属性
        configureRejectionPolicy(element, builder);
        //设置poolSize属性
        String poolSize = element.getAttribute("pool-size");
        if (StringUtils.hasText(poolSize)) {
        
            builder.addPropertyValue("poolSize", poolSize);
        }
    }
    
    private void configureRejectionPolicy(Element element, BeanDefinitionBuilder builder) {
        
        //获取设置的拒绝策略字符串
        String rejectionPolicy = element.getAttribute("rejection-policy");
        //如果没有设置，那么就不配置拒绝策略
        if (!StringUtils.hasText(rejectionPolicy)) {
        
            return;
        }
        //Java提供的4种基本拒绝策略的前缀包路径
        String prefix = "java.util.concurrent.ThreadPoolExecutor.";
        String policyClassName;
        //如果值为ABORT，那么拒绝策略就是AbortPolicy
        if (rejectionPolicy.equals("ABORT")) {
        
            policyClassName = prefix + "AbortPolicy";
        }
        //如果值为CALLER_RUNS，那么拒绝策略就是CallerRunsPolicy
        else if (rejectionPolicy.equals("CALLER_RUNS")) {
        
            policyClassName = prefix + "CallerRunsPolicy";
        }
        //如果值为DISCARD，那么拒绝策略就是DiscardPolicy
        else if (rejectionPolicy.equals("DISCARD")) {
        
            policyClassName = prefix + "DiscardPolicy";
        }
        //如果值为DISCARD_OLDEST，那么拒绝策略就是DiscardOldestPolicy
        else if (rejectionPolicy.equals("DISCARD_OLDEST")) {
        
            policyClassName = prefix + "DiscardOldestPolicy";
        }
        //如果值是其他字符串，那么直接采用该字符串
        else {
        
            policyClassName = rejectionPolicy;
        }
        //设置rejectedExecutionHandler拒绝策略属性
        builder.addPropertyValue("rejectedExecutionHandler", new RootBeanDefinition(policyClassName));
    }

## 2.3 TaskExecutorFactoryBean任务执行器工厂 ##

TaskExecutorFactoryBean是一个任务执行器工厂，用于获取任务执行器！  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center 3]  
  实现了`FactoryBean`接口，它的getObject方法就是用于返回真正的任务执行器实例！还是实现了其他回调接口，用于辅助初始化任务执行器！  
  通过TaskExecutorFactoryBean的源码可知，使用< task:executor/>标签配置的执行器的类型是`org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor`类型。ThreadPoolTaskExecutor内部包装了一个Java的ThreadPoolExecutor，实际的工作仍然是委托内部的ThreadPoolExecutor中的线程！  
  主要方法就是`afterPropertiesSet`回调方法，用于根据给定参数创建ThreadPoolTaskExecutor执行器。

/**
     * 用于创建org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor实例的FactoryBean
     */
    public class TaskExecutorFactoryBean implements
            FactoryBean<TaskExecutor>, BeanNameAware, InitializingBean, DisposableBean {
        
    
        @Nullable
        private String poolSize;
    
        @Nullable
        private Integer queueCapacity;
    
        @Nullable
        private RejectedExecutionHandler rejectedExecutionHandler;
    
        @Nullable
        private Integer keepAliveSeconds;
    
        @Nullable
        private String beanName;
    
        @Nullable
        private ThreadPoolTaskExecutor target;
    
    
        public void setPoolSize(String poolSize) {
        
            this.poolSize = poolSize;
        }
    
        public void setQueueCapacity(int queueCapacity) {
        
            this.queueCapacity = queueCapacity;
        }
    
        public void setRejectedExecutionHandler(RejectedExecutionHandler rejectedExecutionHandler) {
        
            this.rejectedExecutionHandler = rejectedExecutionHandler;
        }
    
        public void setKeepAliveSeconds(int keepAliveSeconds) {
        
            this.keepAliveSeconds = keepAliveSeconds;
        }
    
        @Override
        public void setBeanName(String beanName) {
        
            this.beanName = beanName;
        }
    
        /**
         * 初始化回调方法，在TaskExecutorFactoryBean被实例化之后调用
         * <p>
         * 用于根据给定参数创建执行器
         */
        @Override
        public void afterPropertiesSet() {
        
            //创建一个默认ThreadPoolTaskExecutor执行器
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            /*确定线程数，包括核心线程数与最大线程数*/
            determinePoolSizeRange(executor);
            //设置队列容量，如果存在
            if (this.queueCapacity != null) {
        
                executor.setQueueCapacity(this.queueCapacity);
            }
            //设置超时时间（秒），如果存在
            if (this.keepAliveSeconds != null) {
        
                executor.setKeepAliveSeconds(this.keepAliveSeconds);
            }
            //设置拒绝策略，如果存在
            if (this.rejectedExecutionHandler != null) {
        
                executor.setRejectedExecutionHandler(this.rejectedExecutionHandler);
            }
            //设置线程前缀为beanName，也就是<task:executor/>的id，如果没有指定id将会自动生成beanName
            if (this.beanName != null) {
        
                executor.setThreadNamePrefix(this.beanName + "-");
            }
            /*调用executor的afterPropertiesSet方法初始化执行器，因此我们不必手动调用*/
            executor.afterPropertiesSet();
            //赋值
            this.target = executor;
        }
    
        /**
         * 确定线程数，包括核心线程数与最大线程数
         *
         * @param executor 执行器
         */
        private void determinePoolSizeRange(ThreadPoolTaskExecutor executor) {
        
            //如果设置了poolSize值
            if (StringUtils.hasText(this.poolSize)) {
        
                try {
        
                    int corePoolSize;
                    int maxPoolSize;
                    //首先尝试获取"-"的位置
                    int separatorIndex = this.poolSize.indexOf('-');
                    //如果不为-1，说明poolSize中存在"-"
                    if (separatorIndex != -1) {
        
                        //那么将"-"之前的部分数据解析为corePoolSize，将"-"之后的部分数据解析为maxPoolSize
                        corePoolSize = Integer.parseInt(this.poolSize.substring(0, separatorIndex));
                        maxPoolSize = Integer.parseInt(this.poolSize.substring(separatorIndex + 1, this.poolSize.length()));
                        //如果核心线程数大于最大线程数，那么抛出异常
                        if (corePoolSize > maxPoolSize) {
        
                            throw new IllegalArgumentException(
                                    "Lower bound of pool-size range must not exceed the upper bound");
                        }
                        //如果没有设置队列容量，那么将会采用默认容量
                        //ThreadPoolTaskExecutor默认队列容量为Integer.MAX_VALUE，相当于无界队列
                        if (this.queueCapacity == null) {
        
                            // No queue-capacity provided, so unbounded
                            //并且如果核心线程数设置为0
                            if (corePoolSize == 0) {
        
                                // Actually set 'corePoolSize' to the upper bound of the range
                                // but allow core threads to timeout...
                                //那么允许核心线程超时，并且设置核心线程数等于最大线程数
                                executor.setAllowCoreThreadTimeOut(true);
                                corePoolSize = maxPoolSize;
                            }
                            //否则抛出异常，这说明如果poolSize采用"-"同时配置了核心线程与最大线程数，并且核心线程数不为0
                            //那么queueCapacity必须设置值，因为默认的queueCapacity为Integer.MAX_VALUE，即无界队列
                            else {
        
                                // Non-zero lower bound implies a core-max size range...
                                throw new IllegalArgumentException(
                                        "A non-zero lower bound for the size range requires a queue-capacity value");
                            }
                        }
                    }
                    //如果为-1，说明poolSize中不存在"-"，也就是说只设置了一个值
                    //那么corePoolSize和maxPoolSize都设置为该值
                    else {
        
                        Integer value = Integer.valueOf(this.poolSize);
                        corePoolSize = value;
                        maxPoolSize = value;
                    }
                    executor.setCorePoolSize(corePoolSize);
                    executor.setMaxPoolSize(maxPoolSize);
                } catch (NumberFormatException ex) {
        
                    //如果是非数值类型，那么抛出异常
                    throw new IllegalArgumentException("Invalid pool-size value [" + this.poolSize + "]: only single " +
                            "maximum integer (e.g. \"5\") and minimum-maximum range (e.g. \"3-5\") are supported", ex);
                }
            }
        }
    
        /**
         * 返回对应的ThreadPoolTaskExecutor实例
         */
        @Override
        @Nullable
        public TaskExecutor getObject() {
        
            return this.target;
        }
    
        @Override
        public Class<? extends TaskExecutor> getObjectType() {
        
            return (this.target != null ? this.target.getClass() : ThreadPoolTaskExecutor.class);
        }
    
        @Override
        public boolean isSingleton() {
        
            return true;
        }
    
    
        @Override
        public void destroy() {
        
            //销毁的回调方法，在内部会调用执行器的shutdown方法
            if (this.target != null) {
        
                this.target.destroy();
            }
        }
    }

### 2.3.1 executor.afterPropertiesSet执行器初始化 ###

**在TaskExecutorFactoryBean的afterPropertiesSet方法的最后，会调用executor的afterPropertiesSet方法初始化执行器内部的线程池！**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center 4]  
  源码比较简单，没啥可讲的。唯一需要注意的是，**如果我们要自己配置ThreadPoolTaskExecutor来作为执行器，那么需要注意它的一些默认值，比如它的队列容量默认是Integer.MAX\_VALUE，相当于无界队列了，比如它的最大线程数默认是Integer.MAX\_VALUE，这些属性看看源码就知道了。**

/**
     * ExecutorConfigurationSupport的方法
     */
    @Override
    public void afterPropertiesSet() {
        
        initialize();
    }
    
    
    @Nullable
    private ExecutorService executor;
    
    /**
     * 父类ExecutorConfigurationSupport的方法
     * <p>
     * 设置ExecutorService
     */
    public void initialize() {
        
        if (logger.isInfoEnabled()) {
        
            logger.info("Initializing ExecutorService" + (this.beanName != null ? " '" + this.beanName + "'" : ""));
        }
        //配置线程前缀，可以看到就是beanName
        if (!this.threadNamePrefixSet && this.beanName != null) {
        
            setThreadNamePrefix(this.beanName + "-");
        }
        //初始化真正的执行器，threadFactory就是this自身，主要目的就是用于创建线程时配置线程组、优先级、是否是后台线程等信息
        this.executor = initializeExecutor(this.threadFactory, this.rejectedExecutionHandler);
    }
    
    
    /*ThreadPoolTaskExecutor的属性，提供了一些默认值*/
    
    
    private int corePoolSize = 1;
    
    private int maxPoolSize = Integer.MAX_VALUE;
    
    private int keepAliveSeconds = 60;
    
    private int queueCapacity = Integer.MAX_VALUE;
    
    private boolean allowCoreThreadTimeOut = false;
    
    @Nullable
    private TaskDecorator taskDecorator;
    @Nullable
    private ThreadPoolExecutor threadPoolExecutor;
    
    /**
     * ThreadPoolTaskExecutor的方法
     * <p>
     * 这个方法向其基类公开ExecutorService，但在内部存储实际的ThreadPoolExecutor句柄。
     * 不要为了替换executor而重写此方法，而应只是为了装饰ExecutorService句柄或存储自定义状态。
     */
    @Override
    protected ExecutorService initializeExecutor(
            ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
        
        //根据队列容量创建任务队列
        BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
    
        ThreadPoolExecutor executor;
        //如果任务包装器不为null，一般都是null，但是可以设置
        if (this.taskDecorator != null) {
        
            //构建一个ThreadPoolExecutor，并且使用装饰器装饰即将执行的任务
            executor = new ThreadPoolExecutor(
                    this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
                    queue, threadFactory, rejectedExecutionHandler) {
        
                @Override
                public void execute(Runnable command) {
        
                    Runnable decorated = taskDecorator.decorate(command);
                    if (decorated != command) {
        
                        decoratedTaskMap.put(decorated, command);
                    }
                    super.execute(decorated);
                }
            };
        } else {
        
            //构建一个ThreadPoolExecutor
            executor = new ThreadPoolExecutor(
                    this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
                    queue, threadFactory, rejectedExecutionHandler);
    
        }
        //设置allowCoreThreadTimeOut属性
        if (this.allowCoreThreadTimeOut) {
        
            executor.allowCoreThreadTimeOut(true);
        }
        //赋值
        this.threadPoolExecutor = executor;
        return executor;
    }
    
    
    /**
     * ThreadPoolTaskExecutor的方法
     * <p>
     * 创建用于ThreadPoolExecutor的BlockingQueue
     * 如果配置的容量大于0，则设置LinkedBlockingQueue，否则设置SynchronousQueue
     */
    protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
        
        if (queueCapacity > 0) {
        
            return new LinkedBlockingQueue<>(queueCapacity);
        } else {
        
            return new SynchronousQueue<>();
        }
    }

# 3 @EnableAsync注解解析 #

## 3.1 @EnableAsync注解简介 ##

`@EnableAsync`注解可以替代`<task:annotation-driven/>`标签开启异步任务支持的功能，根据上面`<task:annotation-driven/>`标签的源码分析，我们可以猜测@EnableAsync注解同样会向容器注册一个名为`"org.springframework.context.annotation.internalAsyncAnnotationProcessor"`的`AsyncAnnotationBeanPostProcessor`，用于异步任务的代理对象的创建！  
  此前的文章中我们已经学习过这个注解了，它有如下属性：

<table> 
 <thead> 
  <tr> 
   <th>属性</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Class&lt;? extends Annotation&gt; <code>annotation</code>() default Annotation.class</td> 
   <td>指示要在类或方法级别支持的async注解类型，默认情况下，将支持到 Spring 的@Async注解和 EJB 3.1 @ javax.ejb.Asynchronous注解。此属性的存在，主要是为了方便开发人员可以提供自定义的异步任务注解，以指示应异步调用方法（或给定类的所有方法）。</td> 
  </tr> 
  <tr> 
   <td>boolean <code>proxyTargetClass</code>() default false</td> 
   <td>同&lt; task:annotation-driven/&gt;标签的proxy-target-class属性</td> 
  </tr> 
  <tr> 
   <td>AdviceMode <code>mode</code>() default AdviceMode.PROXY</td> 
   <td>同&lt; task:annotation-driven/&gt;标签的mode属性</td> 
  </tr> 
  <tr> 
   <td>int <code>order</code>() default Ordered.LOWEST_PRECEDENCE</td> 
   <td>AsyncAnnotationBeanPostProcessor的order属性，指示引用该处理器的顺序，默认值Ordered.LOWEST_PRECEDENCE，将会在其他BeanPostProcessor执行完毕之后才执行！主要目的是对于已经是代理对象的bean（被前面的其他后处理器创建了代理对象）可以直接添加一个advisor，而无需创建再次一个双重代理对象。</td> 
  </tr> 
 </tbody> 
</table>

## 3.2 @EnableAsync注解源码 ##

**Spring的@Enable系列注解的原理都差不多，基本上都是基于该注解上的@Import元注解来实现对于某种功能的支持的！@Import注解我们在学习[ConfigurationClassPostProcessor配置类的源码][ConfigurationClassPostProcessor]的时候就讲过了，简单的说该注解同样用于尝试向容器注册bean定义，规则如下：**

1.  对于`普通class`，将会把该class本身注入到容器中；
2.  对于`ImportSelectorBean的class`，将会把selectImports方法返回的一批class注入到容器中。该class本身不会注入到容器中。
    
    1.  源码中还要求排除匹配exclusionFilter过滤器的类名。默认的filter就DEFAULT\_EXCLUSION\_FILTER，该filter的条件是如果类名以"java.lang.annotation."或者"org.springframework.stereotype."开头，就返回true，或者返回false。
3.  对于`ImportBeanDefinitionRegistrarBean的class`，可以通过registerBeanDefinitions方法自定义的注册一批bean定义，后面将会将实例化。该class本身不会注入到容器中。

**对于@EnableAsync上的@Import注解，它指定的是AsyncConfigurationSelector.class，这个类就是一个ImportSelectorBean，因此我们直接看这个类的selectImports方法即可！**  
  **AsyncConfigurationSelector还继承了AdviceModeImportSelector，这是一个通用的基类，该基类及其实现类将会根据AdviceMode选择导入不同的注册的bean！**

/**
     * 这是一个通用基类，用于根据AdviceMode来选择注册不同的bean定义，该基类被用于广泛的支持@Enable*注解
     * 比如EnableAsync、@EnableCaching、@EnableTransactionManagement(需要spring-tx依赖)
     *
     * @param <A> 注解类型泛型
     */
    public abstract class AdviceModeImportSelector<A extends Annotation> implements ImportSelector {
        
    
        /**
         * AdviceMode属性的默认属性名字为"mode"
         */
        public static final String DEFAULT_ADVICE_MODE_ATTRIBUTE_NAME = "mode";
    
    
        /**
         * 获取AdviceMode属性的属性名，默认为"mode"，子类可以覆盖重写
         */
        protected String getAdviceModeAttributeName() {
        
            return DEFAULT_ADVICE_MODE_ATTRIBUTE_NAME;
        }
    
        /**
         * 该方法就是核心方法，在解析@Import注解时就会自动调用
         * <p>
         * 该方法首先会校验指定的注解中是否存在类型为AdviceMode名字为getAdviceModeAttributeName()返回值的属性
         * 然后会调用另一个selectImports(AdviceMode adviceMode)方法，该方法被子类重写，用于根据mode值判断应该使用哪些bean
         */
        @Override
        public final String[] selectImports(AnnotationMetadata importingClassMetadata) {
        
            //获取确定的泛型类型
            Class<?> annType = GenericTypeResolver.resolveTypeArgument(getClass(), AdviceModeImportSelector.class);
            Assert.state(annType != null, "Unresolvable type argument for AdviceModeImportSelector");
            //获取该泛型注解的属性集合
            AnnotationAttributes attributes = AnnotationConfigUtils.attributesFor(importingClassMetadata, annType);
            if (attributes == null) {
        
                throw new IllegalArgumentException(String.format(
                        "@%s is not present on importing class '%s' as expected",
                        annType.getSimpleName(), importingClassMetadata.getClassName()));
            }
            //从属性集合中根据AdviceMode属性的属性名获取属性值
            AdviceMode adviceMode = attributes.getEnum(getAdviceModeAttributeName());
            //调用selectImports(AdviceMode adviceMode)方法选择合适bean，该方法被子类重写
            String[] imports = selectImports(adviceMode);
            //不能为null
            if (imports == null) {
        
                throw new IllegalArgumentException("Unknown AdviceMode: " + adviceMode);
            }
            return imports;
        }
    
        /**
         * 根据给定的AdviceMode确定应导入哪些类，该方法应被子类重写
         * <p>
         * 返回null将会抛出IllegalArgumentException异常.
         */
        @Nullable
        protected abstract String[] selectImports(AdviceMode adviceMode);
    
    }
    
    /**
     * 实现了AdviceModeImportSelector，泛型类型为@EnableAsync注解
     * 重写了selectImports(AdviceMode adviceMode)方法
     */
    public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> {
        
    
        /**
         * 支持Aspectj的静态代理织入
         */
        private static final String ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME =
                "org.springframework.scheduling.aspectj.AspectJAsyncConfiguration";
    
    
        /**
         * 如果EnableAsync#mode是PROXY，那么返回ProxyAsyncConfiguration
         * 如果EnableAsync#mode是ASPECTJ，那么返回AspectJAsyncConfiguration
         */
        @Override
        @Nullable
        public String[] selectImports(AdviceMode adviceMode) {
        
            //判断注解的adviceMode属性值
            switch (adviceMode) {
        
                case PROXY:
                    //一般都是PROXY，因此将会注册ProxyAsyncConfiguration的bean定义，进行动态代理
                    return new String[]{
        ProxyAsyncConfiguration.class.getName()};
                case ASPECTJ:
                    //支持Aspectj的静态代理织入
                    return new String[]{
        ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME};
                default:
                    //其他值则返回null，将会抛出异常
                    return null;
            }
        }
    
    }

**可以看到，默认情况下，将会向容器注册一个`ProxyAsyncConfiguration的bean`定义。**

### 3.2.1 ProxyAsyncConfiguration ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center 5]  
  **ProxyAsyncConfiguration继承了AbstractAsyncConfiguration，该类是一个抽象基配置类，为启用 Spring 的异步方法执行功能提供了通用实现，获取注解的属性以及获取自定义异步任务配置AsyncConfigurer的方法都是在父类中完成的！**

/**
     * 抽象基配置类，为启用 Spring 的异步方法执行功能提供了通用实现。
     * <p>
     * 该类实现了ImportAware感知接口，可以获取当前注解的自身元数据
     */
    @Configuration
    public abstract class AbstractAsyncConfiguration implements ImportAware {
        
    
        @Nullable
        protected AnnotationAttributes enableAsync;
    
        @Nullable
        protected Supplier<Executor> executor;
    
        @Nullable
        protected Supplier<AsyncUncaughtExceptionHandler> exceptionHandler;
    
        /**
         * 该方法被ImportAwareBeanPostProcessor后处理器调用，用于将当前注解自身的元数据设置进来
         */
        @Override
        public void setImportMetadata(AnnotationMetadata importMetadata) {
        
            //获取注解属性设置给enableAsync
            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());
            }
        }
    
        /**
         * 如果beanFactory中还有其他AsyncConfigurer类型的bean（当然可以没有），那么全部注入进来
         * <p>
         * AsyncConfigurer是一个自定义的异步任务配置类，我们可以通过该配置类来配置自定义的任务执行器
         * 在Spring异步任务学习的时候我们就说过该扩展点了
         */
        @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();
            //获取配置类中配置的异步执行器设置给executor属性
            this.executor = configurer::getAsyncExecutor;
            //获取配置类中配置的异常处理器设置给exceptionHandler属性
            this.exceptionHandler = configurer::getAsyncUncaughtExceptionHandler;
        }
    
    }

**ProxyAsyncConfiguration内部仅有一个asyncAdvisor方法，该方法的目的就是解析各种属性，并且添加一个名为"org.springframework.context.annotation.internalAsyncAnnotationProcessor"的AsyncAnnotationBeanPostProcessor后处理器这就是基于XML标签的配置殊途同归了！**

/**
     * TaskManagementConfigUtils的属性
     * <p>
     * AsyncAnnotationBeanPostProcessor的beanName
     */
    public static final String ASYNC_ANNOTATION_PROCESSOR_BEAN_NAME =
            "org.springframework.context.annotation.internalAsyncAnnotationProcessor";
    
    /**
     * 配置类，继承了AbstractAsyncConfiguration
     */
    @Configuration
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public class ProxyAsyncConfiguration extends AbstractAsyncConfiguration {
        
    
        /**
         * 一个@Bean方法，beanName为"org.springframework.context.annotation.internalAsyncAnnotationProcessor"
         */
        @Bean(name = TaskManagementConfigUtils.ASYNC_ANNOTATION_PROCESSOR_BEAN_NAME)
        @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
        public AsyncAnnotationBeanPostProcessor asyncAdvisor() {
        
            //判断是否具有@EnableAsync注解
            Assert.notNull(this.enableAsync, "@EnableAsync annotation metadata was not injected");
            //新建一个AsyncAnnotationBeanPostProcessor实例
            AsyncAnnotationBeanPostProcessor bpp = new AsyncAnnotationBeanPostProcessor();
            //设置从AsyncConfigurer中获取的配置信息，包括执行器和异常处理器
            bpp.configure(this.executor, this.exceptionHandler);
            //获取annotation属性值，这表示自定义的异步任务注解
            Class<? extends Annotation> customAsyncAnnotation = this.enableAsync.getClass("annotation");
            //如果值不等于属性默认值，这说明使用者手动设置了值
            if (customAsyncAnnotation != AnnotationUtils.getDefaultValue(EnableAsync.class, "annotation")) {
        
                //那么将设置的注解作为异步任务注解，并且不会解析默认注解
                bpp.setAsyncAnnotationType(customAsyncAnnotation);
            }
            //设置proxyTargetClass属性，它决定了采用什么样的代理
            bpp.setProxyTargetClass(this.enableAsync.getBoolean("proxyTargetClass"));
            //设置order属性，它决定了后处理执行顺序
            bpp.setOrder(this.enableAsync.<Integer>getNumber("order"));
            return bpp;
        }
    
    }

# 4 小结 #

从源码中能看出来，无论是`<task:annotation-driven/>标签`还是`@EnableAsync注解`，最终的目的就是向容器注入`AsyncAnnotationBeanPostProcessor后处理器`！那么下一步，我们自然是深入了解这个后处理器，下一篇文章我们将详细介绍！

**相关文章：**  
  [https://spring.io/][https_spring.io]  
  [Spring 5.x 学习][Spring 5.x]  
  [Spring 5.x 源码][Spring 5.x 1]

> 如有需要交流，或者文章有误，请直接留言。另外希望点赞、收藏、关注，我将不间断更新各种Java学习博客！

[Spring 5.x _7_Async]: https://blog.csdn.net/weixin_43767015/article/details/110135495
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221024/5b4ff7da135141ffaf8a5101e971798a.png
[Spring IoC_3_parseDefaultElement_parseCustomElement_registerBeanDefinition_Bean]: https://blog.csdn.net/weixin_43767015/article/details/109274050
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221024/192faa6cf7964bb7931a9aa83fc787f6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221024/0edf3d3e873448dea613948b54dbae7f.png
[context_property-placeholder]: https://blog.csdn.net/weixin_43767015/article/details/109459643
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221024/a11ce813519f46e1acf2406b0d13d63f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221024/5ac3862d2b3a407d8e8ba72959eb7a21.png
[ConfigurationClassPostProcessor]: https://blog.csdn.net/weixin_43767015/article/details/109619217
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzc2NzAxNQ_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221024/878e5750aea0441f91275112a181f923.png
[https_spring.io]: https://spring.io/
[Spring 5.x]: https://blog.csdn.net/weixin_43767015/category_10402193.html
[Spring 5.x 1]: https://blog.csdn.net/weixin_43767015/category_10402194.html