首页>>后端>>java->SpringBoot@Async原理解析

SpringBoot@Async原理解析

时间:2023-12-05 本站 点击:0

之前看了一篇文章,里面提到了使用@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);}2

DynamicAdvisedInterceptor这个类有一个AdvisedSupport属性,这个属性其实是从异步注解后置处理器的advisor拿过来,然后执行new CglibMethodInvocation那一行代码,把之前设置在后置处理器中的Advice传进去,接着会跳到AsyncExecutionInterceptor的invoke方法去,把任务放到线程池汇中去执行,然后处理返回值,整个流程就结束了。

总结与启示

整个流程就是,首先在项目启动阶段,通过@EnableAsync注解导入一些配置类,最终实例化一个异步注解的后置处理器(AsyncAnnotationBeanPostProcessor),这个后置处理器有自己的切入点和处理逻辑,会拦截项目中所有的bean,如果某个bean符合该后置处理器的切入点,那么SpringBoot会通过AOP生成一个代理对象,生成代理对象时会设置一个回调,回调的内容就是后置处理器中的处理逻辑(实际逻辑就是将异步方法内容放入线程池中执行),并将这个代理对象注入到使用的地方。

当真正调用异步方法时,因为注入的是代理对象,那么调用到异步方法之前会进入之前设置的回调,去执行异步方法内容,执行完毕后会根据不同的返回值类型处理返回值,至此异步方法就执行完毕了。整篇文章还是有点抽象,过程描述不太详细,只是描述每个类的功能,之前想画一下时序图,但是因为方法调用太多画的图会很复杂,所以强烈建议读者在应用启动阶段和方法调用阶段设置断点,一步步Debug会理解得更深刻。

看完@Async源码感触还是挺多的,首先就是源码的类结构的设计真的很优秀,很多思路可以借鉴,引入到自己的项目中。其次看源码我的一个思路就是,从上往下,从父类到子类,父接口到子接口,对于每一个类,确定是如何实例化的,每个属性是如何设置的,有哪些方法,每个方法的每一行是做什么的,最后大致看完以后,设置断点一步步Debug走下去,边Debug边看每个类的结构,这样对某个类就有个整体把握。

当然源码很复杂,有些地方不可能看的很细,在我看@Async源码的时候有些地方也没有理解,比如AOP那一块,以后有机会再慢慢看看,总之看源码慢慢来,看完有收获就可以。

原文:https://juejin.cn/post/7097521638119309319


本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。
如若转载,请注明出处:/java/12187.html