Spring异步注解@Async

男娘i 2024-03-25 23:57 68阅读 0赞

Spring提供了@Async异步注解，该注解可以被标注在方法上，以便异步地调用该方法。

@Target({ElementType.TYPE, ElementType.METHOD})
        @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
             * the asynchronous operation(s), 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.
             * <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 "";
        }

### @Async的基本应用 ###

// 开启异步支持
    @EnableAsync
    @SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
    public class IntlCommonApplication {
    
        public static void main(String[] args) {
            SpringApplication.run(IntlCommonApplication.class, args);
        }
    
    }
    
    @RestController
    public class TestController {
    
        @Resource
        private AsyncTestService  asyncTestService;
    
        @RequestMapping("/test")
        public String test(){
            asyncTestService.test();
            System.out.println("目标方法执行完了吗？");
            return "OK";
        }
    
    }
    
    @Service
    public class AsyncTestServiceImpl implements AsyncTestService {
        // 标记异步方法
        @Async
        @Override
        public void test() {
            try {
                TimeUnit.SECONDS.sleep(20);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("test:" + this.getClass() + "线程：" + Thread.currentThread().getName());
        }
    }

![05e5f0ea1ff0072ad5b0316e47d29135.png][]

如果是同步执行，那么控制台打印的顺序应该是Service输出在上，Controller输出在下。通过控制台打印的日志，可以证明异步的注解生效了。

### @Async原理解析 ###

test()方法标注了@Async注解，该方法就异步执行了，那么该方法肯定是被拦截了，方法拦截肯定存在一个方法拦截器MethodInterceptor。

@FunctionalInterface
    public interface MethodInterceptor extends Interceptor {
        @Nullable
        Object invoke(@Nonnull MethodInvocation invocation) throws Throwable;
    }

方法拦截器是一个接口，对异步方法的拦截，肯定是该接口的一个实现类。那么如何找到这个实现类呢，我们可以从@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);
    
    }

#### **实例化ProxyAsyncConfiguration** ####

@Configuration(
        proxyBeanMethods = false
    )
    @Role(2)
    public class ProxyAsyncConfiguration extends AbstractAsyncConfiguration {
        public ProxyAsyncConfiguration() {
        }
    
        @Bean(
            name = {"org.springframework.context.annotation.internalAsyncAnnotationProcessor"}
        )
        @Role(2)
        public AsyncAnnotationBeanPostProcessor asyncAdvisor() {
            Assert.notNull(this.enableAsync, "@EnableAsync annotation metadata was not injected");
            // 生明异步注解后置处理器
            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((Integer)this.enableAsync.getNumber("order"));
            return bpp;
        }
    }

ProxyAsyncConfiguration这个类主要是声明了AsyncAnnotationBeanPostProcessor这个Bean，就是异步注解后置处理器，分析其继承体体系，发现其实现了BeanFactoryAware接口，实现该接口的类，spring容器在创建该bean时，会回调：void setBeanFactory(BeanFactory beanFactory) throws BeansException。

![f6b15bb0239c4ca8725a18cd410c59fc.png][]

#### AsyncAnnotationBeanPostProcessor ####

setBeanFactory方法往容器中添加了一个增强器AsyncAnnotationAdvisor

![9c815f10cccd44ceaf95cf9ce4e6a4ac.png][]

#### AsyncAnnotationAdvisor ####

AsyncAnnotationAdvisor的构造器当中执行了buildAdvice增强方法

![9c31677cfd35cfe7ac56e47aee94dace.png][]

buildAdvice方法构建的拦截器

![1c62f56721e78b86843a4ae90f058117.png][]

#### AnnotationAsyncExecutionInterceptor ####

AnnotationAsyncExecutionInterceptor的继承体系

![755d62f4306cd8231bb801c9c9c1412a.png][]

重点看它的父类AsyncExecutionInterceptor

![dafa6379eead68ff1e5758ae66cc4769.png][]

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

接下来看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);
            }
        }
    
    }

![7ca8a3462c124bd815f45723ee07ba37.png][]

### 总结 ###

通过AsyncExecutionAspectSupport源码，我们可以知道线程池会首先通过用户自己配置的为准，找不到就使用默认的线程池，Spring会优先搜索TaskExecutor类型的bean或者名为taskExecutor的Executor类型的bean，如果都不存在的话，会使用SimpleAsyncTaskExecutor（不推荐使用）执行器。所以，我们可以自定义线程池，然后注入spring中，将bean的名字放到@Async注解的value值即可。

![91e25f4346a4888ca122818ec5c4149a.png][]

--------------------

[参考文章--深入解析SpringBoot中@Async注解原理][--_SpringBoot_Async]

[05e5f0ea1ff0072ad5b0316e47d29135.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/577a4f07db1c4acf92f2446af8d16e13.png
[f6b15bb0239c4ca8725a18cd410c59fc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/48b746f9217b45b9b5f317e07d204412.png
[9c815f10cccd44ceaf95cf9ce4e6a4ac.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/d780c4d9dfa846d1acfc13c1b718cbea.png
[9c31677cfd35cfe7ac56e47aee94dace.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/b9e1e755c5b54b3b9fcd1a927c397a0c.png
[1c62f56721e78b86843a4ae90f058117.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/6c34cacab9154a64a1c38b24ba6f0886.png
[755d62f4306cd8231bb801c9c9c1412a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/373a0f6616ed49019af18b7ef58d4c1b.png
[dafa6379eead68ff1e5758ae66cc4769.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/23d81624efe946a89fcfaed39b4757db.png
[7ca8a3462c124bd815f45723ee07ba37.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/31b255b0538f42e4a6dadd7240c5d6b2.png
[91e25f4346a4888ca122818ec5c4149a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/7948e90e3ea74977b03df7c0909d2490.png
[--_SpringBoot_Async]: https://blog.csdn.net/AE86JayChou/article/details/124677305