首页 科技正文

Spring中异步注解@Async的使用、原理及使用时可能导致的问题

admin 1个月前 (12-17) 科技 78 2

前言

实在最近都在研究事务相关的内容，之以是写这么一篇文章是由于前面写了一篇关于循环依赖的文章：

《面试必杀技，讲一讲Spring中的循环依赖》

然后，许多同砚碰着了下面这个问题，添加了Spring提供的一个异步注解@Async循环依赖无法被解决了，下面是一些读者的留言跟群里同砚碰着的问题：

本着讲一个知识点就要讲明了、讲透彻的原则，我决议单独写一篇这样的文章对@Async这个注解做一下详细的先容，这个注解带来的问题远远不止循环依赖这么简朴，若是对它不够熟悉的话建议慎用。

文章要点

@Async的基本使用

这个注解的作用在于可以让被标注的方式异步执行，然则有两个前提条件

  1. 设置类上添加@EnableAsync注解
  2. 需要异步执行的方式的所在类由Spring治理
  3. 需要异步执行的方式上添加了@Async注解

我们通过一个Demo体会下这个注解的作用吧

第一步，设置类上开启异步：

@EnableAsync
@Configuration
@ComponentScan("com.dmz.spring.async")
public class Config {

}

第二步，

@Component  // 这个类自己要被Spring治理
public class DmzAsyncService {
    
	@Async  // 添加注解示意这个方式要异步执行
	public void testAsync(){
		try {
			TimeUnit.SECONDS.sleep(1);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("testAsync invoked");
	}
}

第三步，测试异步执行

public class Main {
	public static void main(String[] args) {
		AnnotationConfigApplicationContext ac = new AnnotationConfigApplicationContext(Config.class);
		DmzAsyncService bean = ac.getBean(DmzAsyncService.class);
		bean.testAsync();
		System.out.println("main函数执行完成");
	}
}
// 程序执行效果如下：
// main函数执行完成
// testAsync invoked

通过上面的例子我们可以发现，DmzAsyncService中的testAsync方式是异步执行的，那么这背后的原理是什么呢？我们接着剖析

原理剖析

我们在剖析某一个手艺的时刻，最主要的事情是，一定一定要找到代码的入口，像Spring这种都很显著，入口必定是在@EnableAsync这个注解上面，我们来看看这个注解干了啥事（本文基于5.2.x版本）

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
// 这里是重点，导入了一个ImportSelector
@Import(AsyncConfigurationSelector.class)
public @interface EnableAsync {
    
    // 这个设置可以让程序员设置需要被检查的注解，默认情况下检查的就是@Async注解
	Class<? extends Annotation> annotation() default Annotation.class;
	
    // 默认使用jdk署理
	boolean proxyTargetClass() default false;
	
    // 默认使用Spring AOP
	AdviceMode mode() default AdviceMode.PROXY;
	
    // 在后续剖析我们会发现，这个注解现实往容器中添加了一个
    // AsyncAnnotationBeanPostProcessor，这个后置处置器实现了Ordered接口
    // 这个设置主要代表了AsyncAnnotationBeanPostProcessor执行的顺序
	int order() default Ordered.LOWEST_PRECEDENCE;
}

上面这个注解做的最主要的事情就是导入了一个AsyncConfigurationSelector，这个类的源码如下：

public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> {

	private static final String ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME =
			"org.springframework.scheduling.aspectj.AspectJAsyncConfiguration";

	@Override
	@Nullable
	public String[] selectImports(AdviceMode adviceMode) {
		switch (adviceMode) {
                // 默认会使用SpringAOP举行署理
			case PROXY:
				return new String[] {ProxyAsyncConfiguration.class.getName()};
			case ASPECTJ:
				return new String[] {ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME};
			default:
				return null;
		}
	}

}

这个类的作用是像容器中注册了一个ProxyAsyncConfiguration，这个类的继续关系如下：

我们先看下它的父类AbstractAsyncConfiguration，其源码如下：

@Configuration
public abstract class AbstractAsyncConfiguration implements ImportAware {
	
	@Nullable
	protected AnnotationAttributes enableAsync;

	@Nullable
	protected Supplier<Executor> executor;

	@Nullable
	protected Supplier<AsyncUncaughtExceptionHandler> exceptionHandler;
	
    // 这里主要就是检查将其导入的类上是否有EnableAsync注解
    // 若是没有的话就报错
	@Override
	public void setImportMetadata(AnnotationMetadata importMetadata) {
		this.enableAsync = AnnotationAttributes.fromMap(
				importMetadata.getAnnotationAttributes(EnableAsync.class.getName(), false));
		if (this.enableAsync == null) {
			throw new IllegalArgumentException(
					"@EnableAsync is not present on importing class " + importMetadata.getClassName());
		}
	}
    
    // 将容器中设置的AsyncConfigurer注入
    // 异步执行嘛，以是我们可以设置使用的线程池
    // 另外也可以设置异常处置器
	@Autowired(required = false)
	void setConfigurers(Collection<AsyncConfigurer> configurers) {
		if (CollectionUtils.isEmpty(configurers)) {
			return;
		}
		if (configurers.size() > 1) {
			throw new IllegalStateException("Only one AsyncConfigurer may exist");
		}
		AsyncConfigurer configurer = configurers.iterator().next();
		this.executor = configurer::getAsyncExecutor;
		this.exceptionHandler = configurer::getAsyncUncaughtExceptionHandler;
	}

}

再来看看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() {
		AsyncAnnotationBeanPostProcessor bpp = new AsyncAnnotationBeanPostProcessor();
        // 将通过AsyncConfigurer设置好的线程池跟异常处置器设置到这个后置处置器中
        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(this.enableAsync.<Integer>getNumber("order"));
		return bpp;
	}

}

这个类自己是一个设置类，它的作用是向容器中添加一个AsyncAnnotationBeanPostProcessor。到这一步我们基本上就可以明了了，@Async注解的就是通过AsyncAnnotationBeanPostProcessor这个后置处置器天生一个署理工具来实现异步的，接下来我们就详细看看AsyncAnnotationBeanPostProcessor是若何天生署理工具的，我们主要关注一下几点即可：

  1. 是在生命周期的哪一步完成的署理？
  2. 切点的逻辑是怎么样的？它会对什么样的类举行阻挡？
  3. 通知的逻辑是怎么样的？是若何实现异步的？

基于上面几个问题，我们举行逐一剖析

是在生命周期的哪一步完成的署理？

我们捉住重点，AsyncAnnotationBeanPostProcessor是一个后置处置器器，凭据我们对Spring的领会，大概率是在这个后置处置器的postProcessAfterInitialization方式中完成了署理，直接定位到这个方式，这个方式位于父类AbstractAdvisingBeanPostProcessor中，详细代码如下：

public Object postProcessAfterInitialization(Object bean, String beanName) {
    // 没有通知，或者是AOP的基础设施类，那么不举行署理
    if (this.advisor == null || bean instanceof AopInfrastructureBean) {
        return bean;
    }
	
    // 对已经被署理的类，不再天生署理，只是将通知添加到署理类的逻辑中
    // 这里通过beforeExistingAdvisors决议是将通知添加到所有通知之前照样添加到所有通知之后
    // 在使用@Async注解的时刻，beforeExistingAdvisors被设置成了true
    // 意味着整个方式及其阻挡逻辑都市异步执行
    if (bean instanceof Advised) {
        Advised advised = (Advised) bean;
        if (!advised.isFrozen() && isEligible(AopUtils.getTargetClass(bean))) {
            if (this.beforeExistingAdvisors) {
                advised.addAdvisor(0, this.advisor);
            }
            else {
                advised.addAdvisor(this.advisor);
            }
            return bean;
        }
    }
	
    // 判断需要对哪些Bean举行来署理
    if (isEligible(bean, beanName)) {
        ProxyFactory proxyFactory = prepareProxyFactory(bean, beanName);
        if (!proxyFactory.isProxyTargetClass()) {
            evaluateProxyInterfaces(bean.getClass(), proxyFactory);
        }
        proxyFactory.addAdvisor(this.advisor);
        customizeProxyFactory(proxyFactory);
        return proxyFactory.getProxy(getProxyClassLoader());
    }
    return bean;
}

果不其然，确实是在这个方式中完成的署理。接着我们就要思索，切点的过滤规则是什么呢？

切点的逻辑是怎么样的？

实在也不难猜到一定就是类上添加了@Async注解或者类中含有被@Async注解修饰的方式。基于此，我们看看这个isEligible这个方式的实现逻辑，这个方位位于AbstractBeanFactoryAwareAdvisingPostProcessor中，也是AsyncAnnotationBeanPostProcessor的父类，对应代码如下：

// AbstractBeanFactoryAwareAdvisingPostProcessor的isEligible方式
// 挪用了父类
protected boolean isEligible(Object bean, String beanName) {
    return (!AutoProxyUtils.isOriginalInstance(beanName, bean.getClass()) &&
            super.isEligible(bean, beanName));
}

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;
    }
    // 这里完成的判断
    eligible = AopUtils.canApply(this.advisor, targetClass);
    this.eligibleBeans.put(targetClass, eligible);
    return eligible;
}

现实上最后就是凭据advisor来确定是否要举行署理，在Spring中AOP相关的API及源码剖析，原来AOP是这样子的这篇文章中我们提到过，advisor现实就是一个绑定了切点的通知，那么AsyncAnnotationBeanPostProcessor这个advisor是什么时刻被初始化的呢？我们直接定位到AsyncAnnotationBeanPostProcessor的setBeanFactory方式，其源码如下：

public void setBeanFactory(BeanFactory beanFactory) {
    super.setBeanFactory(beanFactory);
	
    // 在这里new了一个AsyncAnnotationAdvisor
    AsyncAnnotationAdvisor advisor = new AsyncAnnotationAdvisor(this.executor, this.exceptionHandler);
    if (this.asyncAnnotationType != null) {
        advisor.setAsyncAnnotationType(this.asyncAnnotationType);
    }
    advisor.setBeanFactory(beanFactory);
    // 完成了初始化
    this.advisor = advisor;
}

我们来看看AsyncAnnotationAdvisor中的切点匹配规程是怎么样的，直接定位到这个类的buildPointcut方式中，其源码如下：

protected Pointcut buildPointcut(Set<Class<? extends Annotation>> asyncAnnotationTypes) {
    ComposablePointcut result = null;
    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);
}

代码很简朴，就是凭据cpc跟mpc两个匹配器来举行匹配的，第一个是检查类上是否有@Async注解，第二个是检查方式是是否有@Async注解。

那么，到现在为止，我们已经知道了它在何时建立署理，会为什么工具建立署理，最后我们还需要解决一个问题，署理的逻辑是怎么样的，异步到底是若何实现的？

通知的逻辑是怎么样的？是若何实现异步的？

前面也提到了advisor是一个绑定了切点的通知，前面剖析了它的切点，那么现在我们就来看看它的通知逻辑，直接定位到AsyncAnnotationAdvisor中的buildAdvice方式，源码如下：

protected Advice buildAdvice(
    @Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) {

    AnnotationAsyncExecutionInterceptor interceptor = new AnnotationAsyncExecutionInterceptor(null);
    interceptor.configure(executor, exceptionHandler);
    return interceptor;
}

简朴吧，加了一个阻挡器而已，对于interceptor类型的工具，我们关注它的焦点方式invoke就行了，代码如下：

public Object invoke(final MethodInvocation invocation) throws Throwable {
    Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
    Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
    final Method userDeclaredMethod = BridgeMethodResolver.findBridgedMethod(specificMethod);
	
    // 异步执行嘛，先获取到一个线程池
    AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod);
    if (executor == null) {
        throw new IllegalStateException(
            "No executor specified and no default executor set on AsyncExecutionInterceptor either");
    }
	
    // 然后将这个方式封装成一个 Callable工具传入到线程池中执行
    Callable<Object> task = () -> {
        try {
            Object result = invocation.proceed();
            if (result instanceof Future) {
                return ((Future<?>) result).get();
            }
        }
        catch (ExecutionException ex) {
            handleError(ex.getCause(), userDeclaredMethod, invocation.getArguments());
        }
        catch (Throwable ex) {
            handleError(ex, userDeclaredMethod, invocation.getArguments());
        }
        return null;
    };
	// 将义务提交到线程池
    return doSubmit(task, executor, invocation.getMethod().getReturnType());
}

导致的问题及解决方案

问题1：循环依赖报错

就像在这张图里这个读者问的问题，

分为两点回覆：

第一：循环依赖为什么不能被解决？

这个问题实在很简朴，在《面试必杀技，讲一讲Spring中的循环依赖》这篇文章中我从两个方面剖析了循环依赖的处置流程

  1. 简朴工具间的循环依赖处置
  2. AOP工具间的循环依赖处置

凭据这种思绪，@Async注解导致的循环依赖应该属于AOP工具间的循环依赖,也应该能被处置。然则，重点来了，解决AOP工具间循环依赖的焦点方式是三级缓存，如下：

在三级缓存缓存了一个工厂工具，这个工厂工具会挪用getEarlyBeanReference方式来获取一个早期的署理工具的引用，其源码如下：

protected Object getEarlyBeanReference(String beanName, RootBeanDefinition mbd, Object bean) {
   Object exposedObject = bean;
   if (!mbd.isSynthetic() && hasInstantiationAwareBeanPostProcessors()) {
      for (BeanPostProcessor bp : getBeanPostProcessors()) {
          // 看到这个判断了吗，通过@EnableAsync导入的后置处置器
          // AsyncAnnotationBeanPostProcessor基本就不是一个SmartInstantiationAwareBeanPostProcessor
          // 这就意味着纵然我们通过AsyncAnnotationBeanPostProcessor建立了一个署理工具
          // 然则早期露出出去的用于给其余Bean举行注入的谁人工具照样原始工具
         if (bp instanceof SmartInstantiationAwareBeanPostProcessor) {
            SmartInstantiationAwareBeanPostProcessor ibp = (SmartInstantiationAwareBeanPostProcessor) bp;
            exposedObject = ibp.getEarlyBeanReference(exposedObject, beanName);
         }
      }
   }
   return exposedObject;
}

看完上面的代码循环依赖的问题就很显著了，由于早期露出的工具跟最终放入容器中的工具不是同一个，以是报错了。报错的详细位置我在你知道Spring是怎么将AOP应用到Bean的生命周期中的吗? 文章末尾已经剖析过了，本文不再赘述

解决方案

就以上面读者给出的Demo为例，只需要在为B注入A时添加一个@Lazy注解即可

@Component
public class B implements BService {
	
    @Autowired
	@Lazy
	private A a;

	public void doSomething() {
	}
}

这个注解的作用在于，当为B注入A时，会为A天生一个署理工具注入到B中，当真正挪用署理工具的方式时，底层会挪用getBean(a)去建立A工具，然后挪用方式，这个注解的处置时机是在org.springframework.beans.factory.support.DefaultListableBeanFactory#resolveDependency方式中，处置这个注解的代码位于org.springframework.context.annotation.ContextAnnotationAutowireCandidateResolver#buildLazyResolutionProxy，这些代码实在都在我之前的文章中剖析过了

《Spring杂谈 | Spring中的AutowireCandidateResolver》

《谈谈Spring中的工具跟Bean，你知道Spring怎么建立工具的吗？》

以是本文不再做详细剖析

问题2：默认线程池不会复用线程

我以为这是这个注解最坑的地方，没有之一！我们来看看它默认使用的线程池是哪个，在前文的源码剖析中，我们可以看到决议要使用线程池的方式是org.springframework.aop.interceptor.AsyncExecutionAspectSupport#determineAsyncExecutor。其源码如下：

protected AsyncTaskExecutor determineAsyncExecutor(Method method) {
    AsyncTaskExecutor executor = this.executors.get(method);
    if (executor == null) {
        Executor targetExecutor;
        // 可以在@Async注解中设置线程池的名字
        String qualifier = getExecutorQualifier(method);
        if (StringUtils.hasLength(qualifier)) {
            targetExecutor = findQualifiedExecutor(this.beanFactory, qualifier);
        }
        else {
            // 获取默认的线程池
            targetExecutor = this.defaultExecutor.get();
        }
        if (targetExecutor == null) {
            return null;
        }
        executor = (targetExecutor instanceof AsyncListenableTaskExecutor ?
                    (AsyncListenableTaskExecutor) targetExecutor : new TaskExecutorAdapter(targetExecutor));
        this.executors.put(method, executor);
    }
    return executor;
}

最终会挪用到org.springframework.aop.interceptor.AsyncExecutionInterceptor#getDefaultExecutor这个方式中

protected Executor getDefaultExecutor(@Nullable BeanFactory beanFactory) {
   Executor defaultExecutor = super.getDefaultExecutor(beanFactory);
   return (defaultExecutor != null ? defaultExecutor : new SimpleAsyncTaskExecutor());
}

可以看到，它默认使用的线程池是SimpleAsyncTaskExecutor。我们不看这个类的源码，只看它上面的文档注释，如下：

主要说了三点

  1. 为每个义务新起一个线程
  2. 默认线程数不做限制
  3. 不复用线程

就这三点，你还敢用吗？只要你的义务耗时长一点，说不定服务器就给你来个OOM。

解决方案

最好的设施就是使用自定义的线程池，主要有这么几种设置方式

  1. 在之前的源码剖析中，我们可以知道，可以通过AsyncConfigurer来设置使用的线程池

如下：

public class DmzAsyncConfigurer implements AsyncConfigurer {
   @Override
   public Executor getAsyncExecutor() {
      // 建立自定义的线程池
   }
}
  1. 直接在@Async注解中设置要使用的线程池的名称

如下：

public class A implements AService {
	
	private B b;

	@Autowired
	public void setB(B b) {
		System.out.println(b);
		this.b = b;
	}

	@Async("dmzExecutor")
	public void doSomething() {
	}
}

@EnableAsync
@Configuration
@ComponentScan("com.dmz.spring.async")
@Aspect
public class Config {
	@Bean("dmzExecutor")
	public Executor executor(){
		// 建立自定义的线程池
		return executor;
	}
}

总结

本文主要先容了Spring中异步注解的使用、原理及可能碰着的问题，针对每个问题文中也给出了方案。希望通过这篇文章能辅助你彻底掌握@Async注解的使用，知其然并知其以是然！

文章有帮到你的话，记得点个赞哈~
若是本文对你由辅助的话，记得点个赞吧！也迎接关注我的民众号，微信搜索：程序员DMZ，或者扫描下方二维码，随着我一起认认真真学Java,踏踏实实做一个coder。

我叫DMZ，一个在学习路上匍匐前行的小菜鸟！

,

www.px111.net

欢迎进入平心在线官网（原诚信在线、阳光在线）。平心在线官网www.px111.net开放平心在线会员登录网址、平心在线代理后台网址、平心在线APP下载、平心在线电脑客户端下载、平心在线企业邮局等业务。

申博声明:该文看法仅代表作者自己，与本平台无关。转载请注明：Spring中异步注解@Async的使用、原理及使用时可能导致的问题

相关推荐

网友评论

  • (*)

最新评论

  • Allbet币游 2020-12-17 00:03:53 回复

    欧博注册网址www.allbet6.com欢迎进入欧博网址（Allbet Gaming），欧博网址开放会员注册、代理开户、电脑客户端下载、苹果安卓下载等业务。看得很开心呢

    1
  • UG环球官网 2021-01-12 00:01:39 回复

    皇冠APP（www.huangguan.us）是一个提供皇冠代理APP下载、皇冠会员APP下载、皇冠体育最新登录线路、新2皇冠网址的的体育平台。也只有皇冠APP可以真正地带给你顶级体育赛事的娱乐体验感。立马一键皇冠体育开户，世界体育赛事等你欣赏。还行呢，不拖

    2

热门文章

最近发表

标签列表