我們通過一個(gè)Demo體會(huì)下這個(gè)注解的作用吧
第一步，配置類上開啟異步：
@EnableAsync
@Configuration
@ComponentScan("com.dmz.spring.async")
public class Config {
}
第二步，
@Component? // 這個(gè)類本身要被Spring管理
public class DmzAsyncService {
@Async? // 添加注解表示這個(gè)方法要異步執(zhí)行
public void testAsync(){
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("testAsync invoked");
}
}
第三步辟犀，測(cè)試異步執(zhí)行
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函數(shù)執(zhí)行完成");
}
}
// 程序執(zhí)行結(jié)果如下：
// main函數(shù)執(zhí)行完成
// testAsync invoked
通過上面的例子我們可以發(fā)現(xiàn)，DmzAsyncService中的testAsync方法是異步執(zhí)行的绸硕，那么這背后的原理是什么呢堂竟？我們接著分析
原理分析
我們?cè)诜治瞿骋粋€(gè)技術(shù)的時(shí)候，最重要的事情是玻佩，一定一定要找到代碼的入口出嘹，像Spring這種都很明顯，入口必定是在@EnableAsync這個(gè)注解上面咬崔，我們來看看這個(gè)注解干了啥事（本文基于5.2.x版本）
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
// 這里是重點(diǎn)税稼，導(dǎo)入了一個(gè)ImportSelector
@Import(AsyncConfigurationSelector.class)
public @interface EnableAsync {
? ? // 這個(gè)配置可以讓程序員配置需要被檢查的注解，默認(rèn)情況下檢查的就是@Async注解
Class<? extends Annotation> annotation() default Annotation.class;
? ? // 默認(rèn)使用jdk代理
boolean proxyTargetClass() default false;
? ? // 默認(rèn)使用Spring AOP
AdviceMode mode() default AdviceMode.PROXY;
? ? // 在后續(xù)分析我們會(huì)發(fā)現(xiàn)垮斯，這個(gè)注解實(shí)際往容器中添加了一個(gè)
? ? // AsyncAnnotationBeanPostProcessor郎仆，這個(gè)后置處理器實(shí)現(xiàn)了Ordered接口
? ? // 這個(gè)配置主要代表了AsyncAnnotationBeanPostProcessor執(zhí)行的順序
int order() default Ordered.LOWEST_PRECEDENCE;
}
上面這個(gè)注解做的最重要的事情就是導(dǎo)入了一個(gè)AsyncConfigurationSelector，這個(gè)類的源碼如下：
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) {
? ? ? ? ? ? ? ? // 默認(rèn)會(huì)使用SpringAOP進(jìn)行代理
case PROXY:
return new String[] {ProxyAsyncConfiguration.class.getName()};
case ASPECTJ:
return new String[] {ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME};
default:
return null;
}
}
}
這個(gè)類的作用是像容器中注冊(cè)了一個(gè)ProxyAsyncConfiguration甚脉，這個(gè)類的繼承關(guān)系如下：
image-20200719220316319
我們先看下它的父類AbstractAsyncConfiguration丸升，其源碼如下：
@Configuration
public abstract class AbstractAsyncConfiguration implements ImportAware {
@Nullable
protected AnnotationAttributes enableAsync;
@Nullable
protected Supplier<Executor> executor;
@Nullable
protected Supplier<AsyncUncaughtExceptionHandler> exceptionHandler;
? ? // 這里主要就是檢查將其導(dǎo)入的類上是否有EnableAsync注解
? ? // 如果沒有的話就報(bào)錯(cuò)
@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注入
? ? // 異步執(zhí)行嘛，所以我們可以配置使用的線程池
? ? // 另外也可以配置異常處理器
@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這個(gè)類的源碼
@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配置好的線程池跟異常處理器設(shè)置到這個(gè)后置處理器中
? ? ? ? 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;
}
}
這個(gè)類本身是一個(gè)配置類牺氨，它的作用是向容器中添加一個(gè)AsyncAnnotationBeanPostProcessor狡耻。到這一步我們基本上就可以明白了墩剖，@Async注解的就是通過AsyncAnnotationBeanPostProcessor這個(gè)后置處理器生成一個(gè)代理對(duì)象來實(shí)現(xiàn)異步的，接下來我們就具體看看AsyncAnnotationBeanPostProcessor是如何生成代理對(duì)象的夷狰，我們主要關(guān)注一下幾點(diǎn)即可：
是在生命周期的哪一步完成的代理岭皂？
切點(diǎn)的邏輯是怎么樣的？它會(huì)對(duì)什么樣的類進(jìn)行攔截沼头？
通知的邏輯是怎么樣的爷绘？是如何實(shí)現(xiàn)異步的？
基于上面幾個(gè)問題进倍，我們進(jìn)行逐一分析
是在生命周期的哪一步完成的代理土至？
我們抓住重點(diǎn)，AsyncAnnotationBeanPostProcessor是一個(gè)后置處理器器猾昆，按照我們對(duì)Spring的了解陶因，大概率是在這個(gè)后置處理器的postProcessAfterInitialization方法中完成了代理，直接定位到這個(gè)方法垂蜗，這個(gè)方法位于父類AbstractAdvisingBeanPostProcessor中楷扬，具體代碼如下：
public Object postProcessAfterInitialization(Object bean, String beanName) {
? ? // 沒有通知，或者是AOP的基礎(chǔ)設(shè)施類贴见，那么不進(jìn)行代理
? ? if (this.advisor == null || bean instanceof AopInfrastructureBean) {
? ? ? ? return bean;
? ? }
? ? // 對(duì)已經(jīng)被代理的類烘苹，不再生成代理，只是將通知添加到代理類的邏輯中
? ? // 這里通過beforeExistingAdvisors決定是將通知添加到所有通知之前還是添加到所有通知之后
? ? // 在使用@Async注解的時(shí)候片部，beforeExistingAdvisors被設(shè)置成了true
? ? // 意味著整個(gè)方法及其攔截邏輯都會(huì)異步執(zhí)行
? ? 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;
? ? ? ? }
? ? }
? ? // 判斷需要對(duì)哪些Bean進(jìn)行來代理
? ? 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;
}
果不其然镣衡，確實(shí)是在這個(gè)方法中完成的代理。接著我們就要思考吞琐，切點(diǎn)的過濾規(guī)則是什么呢捆探？
切點(diǎn)的邏輯是怎么樣的？
其實(shí)也不難猜到肯定就是類上添加了@Async注解或者類中含有被@Async注解修飾的方法站粟∈蛲迹基于此，我們看看這個(gè)isEligible這個(gè)方法的實(shí)現(xiàn)邏輯奴烙，這個(gè)方位位于AbstractBeanFactoryAwareAdvisingPostProcessor中助被，也是AsyncAnnotationBeanPostProcessor的父類，對(duì)應(yīng)代碼如下：
// AbstractBeanFactoryAwareAdvisingPostProcessor的isEligible方法
// 調(diào)用了父類
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;
}
實(shí)際上最后就是根據(jù)advisor來確定是否要進(jìn)行代理切诀，在Spring中AOP相關(guān)的API及源碼解析揩环，原來AOP是這樣子的這篇文章中我們提到過，advisor實(shí)際就是一個(gè)綁定了切點(diǎn)的通知幅虑，那么AsyncAnnotationBeanPostProcessor這個(gè)advisor是什么時(shí)候被初始化的呢丰滑？我們直接定位到AsyncAnnotationBeanPostProcessor的setBeanFactory方法，其源碼如下：
public void setBeanFactory(BeanFactory beanFactory) {
? ? super.setBeanFactory(beanFactory);
? ? // 在這里new了一個(gè)AsyncAnnotationAdvisor
? ? AsyncAnnotationAdvisor advisor = new AsyncAnnotationAdvisor(this.executor, this.exceptionHandler);
? ? if (this.asyncAnnotationType != null) {
? ? ? ? advisor.setAsyncAnnotationType(this.asyncAnnotationType);
? ? }
? ? advisor.setBeanFactory(beanFactory);
? ? // 完成了初始化
? ? this.advisor = advisor;
}
我們來看看AsyncAnnotationAdvisor中的切點(diǎn)匹配規(guī)程是怎么樣的倒庵，直接定位到這個(gè)類的buildPointcut方法中褒墨，其源碼如下：
protected Pointcut buildPointcut(Set<Class<? extends Annotation>> asyncAnnotationTypes) {
? ? ComposablePointcut result = null;
? ? for (Class<? extends Annotation> asyncAnnotationType : asyncAnnotationTypes) {
? ? ? ? // 就是根據(jù)這兩個(gè)匹配器進(jìn)行匹配的
? ? ? ? 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);
}
代碼很簡(jiǎn)單炫刷，就是根據(jù)cpc跟mpc兩個(gè)匹配器來進(jìn)行匹配的，第一個(gè)是檢查類上是否有@Async注解郁妈，第二個(gè)是檢查方法是是否有@Async注解浑玛。
那么，到現(xiàn)在為止噩咪，我們已經(jīng)知道了它在何時(shí)創(chuàng)建代理顾彰，會(huì)為什么對(duì)象創(chuàng)建代理，最后我們還需要解決一個(gè)問題胃碾，代理的邏輯是怎么樣的涨享，異步到底是如何實(shí)現(xiàn)的？
通知的邏輯是怎么樣的仆百？是如何實(shí)現(xiàn)異步的灰伟？
前面也提到了advisor是一個(gè)綁定了切點(diǎn)的通知，前面分析了它的切點(diǎn)儒旬，那么現(xiàn)在我們就來看看它的通知邏輯，直接定位到AsyncAnnotationAdvisor中的buildAdvice方法帖族，源碼如下：
protected Advice buildAdvice(
? ? @Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) {
? ? AnnotationAsyncExecutionInterceptor interceptor = new AnnotationAsyncExecutionInterceptor(null);
? ? interceptor.configure(executor, exceptionHandler);
? ? return interceptor;
}
簡(jiǎn)單吧栈源，加了一個(gè)攔截器而已，對(duì)于interceptor類型的對(duì)象竖般，我們關(guān)注它的核心方法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);
? ? // 異步執(zhí)行嘛，先獲取到一個(gè)線程池
? ? AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod);
? ? if (executor == null) {
? ? ? ? throw new IllegalStateException(
? ? ? ? ? ? "No executor specified and no default executor set on AsyncExecutionInterceptor either");
? ? }
? ? // 然后將這個(gè)方法封裝成一個(gè) Callable對(duì)象傳入到線程池中執(zhí)行
? ? 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;
? ? };
// 將任務(wù)提交到線程池
? ? return doSubmit(task, executor, invocation.getMethod().getReturnType());
}
深圳網(wǎng)站建設(shè)www.sz886.com
