前言
用過Spring的人多多少少也都用過@Async注解，至于作用嘛，看注解名，大概能猜出來，就是在方法執(zhí)行的時候進行異步執(zhí)行。
一、如何使用@Async
使用@Async注解主要分兩步：
1.在配置類上添加@EnableAsync注解
|
1
2
3
4
5
|
@ComponentScan(value = "com.wang")@Configuration@EnableAsyncpublic class AppConfig {} |
2.在想要異步執(zhí)行的方法上面加上@Async
|
1
2
3
4
5
6
7
8
9
10
11
12
|
@Servicepublic class CycleService2 { @Autowired private CycleService1 cycleService1; @Async public void alsoDo() { System.out.println("create cycleService2"); }} |
二、源碼解讀
1.@EnableAsync的作用
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
|
@Target(ElementType.TYPE)@Retention(RetentionPolicy.RUNTIME)@Documented@Import(AsyncConfigurationSelector.class)public @interface EnableAsync { /** * Indicate the 'async' annotation type to be detected at either class * or method level. * <p>By default, both Spring's @{@link Async} annotation and the EJB 3.1 * {@code @javax.ejb.Asynchronous} annotation will be detected. * <p>This attribute exists so that developers can provide their own * custom annotation type to indicate that a method (or all methods of * a given class) should be invoked asynchronously. * 此處說明的是方法執(zhí)行變成異步，掃描的是哪個注解，目前默認的是Async和Asynchronous，開發(fā)者也可以自定義 */ Class<? extends Annotation> annotation() default Annotation.class; /** * Indicate whether subclass-based (CGLIB) proxies are to be created as opposed * to standard Java interface-based proxies. * <p><strong>Applicable only if the {@link #mode} is set to {@link AdviceMode#PROXY}</strong>. * <p>The default is {@code false}. * <p>Note that setting this attribute to {@code true} will affect <em>all</em> * Spring-managed beans requiring proxying, not just those marked with {@code @Async}. * For example, other beans marked with Spring's {@code @Transactional} annotation * will be upgraded to subclass proxying at the same time. This approach has no * negative impact in practice unless one is explicitly expecting one type of proxy * vs. another — for example, in tests. * 如何proxyTargetClass被設置成true，那么spring的所有proxy都會通過CGLIB方式實現(xiàn)，不再使用Java默認的基于接口的代理實現(xiàn)方式；而且此處如果設置，不僅僅是會影響添加了@Async注解的類的proxy方式，加了@Transactional的類也會變成CGLIB代理，不推薦修改；這個注解只有mode是默認的PROXY，才有意義 */ boolean proxyTargetClass() default false; /** * Indicate how async advice should be applied. * <p><b>The default is {@link AdviceMode#PROXY}.</b> * Please note that proxy mode allows for interception of calls through the proxy * only. Local calls within the same class cannot get intercepted that way; an * {@link Async} annotation on such a method within a local call will be ignored * since Spring's interceptor does not even kick in for such a runtime scenario. * For a more advanced mode of interception, consider switching this to * {@link AdviceMode#ASPECTJ}. * 代理方式的不同，默認的是使用Spring的proxy方式，也可以換成原生的AspectJ的proxy方式。 * 這兩個的區(qū)別作用還是很明顯的 */ AdviceMode mode() default AdviceMode.PROXY; /** * Indicate the order in which the {@link AsyncAnnotationBeanPostProcessor} * should be applied. * <p>The default is {@link Ordered#LOWEST_PRECEDENCE} in order to run * after all other post-processors, so that it can add an advisor to * existing proxies rather than double-proxy. * 因為在beanPostProcessor執(zhí)行的時候，會根據(jù)order值進行排序，此處設置為最低值，就是想讓其最后執(zhí)行 * 其實即使不設置這個值，因為AsyncAnnotationBeanPostProcessor繼承了ProxyProcessorSupport，ProxyProcessorSupport中的order默認也是最小優(yōu)先級 * */ int order() default Ordered.LOWEST_PRECEDENCE;} |
2. AsyncConfigurationSelector的作用
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> { private static final String ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME = "org.springframework.scheduling.aspectj.AspectJAsyncConfiguration"; /** * Returns {@link ProxyAsyncConfiguration} or {@code AspectJAsyncConfiguration} * for {@code PROXY} and {@code ASPECTJ} values of {@link EnableAsync#mode()}, * respectively. */ @Override @Nullable 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; } }} |
看過我之前博客的同學應該知道，其實此處就是往Spring容器中增加一個新的需要掃描的類，很明顯可以看到差別主要集中在adviceMode的差別上。
3. adviceMode:PROXY(默認值)
引入了ProxyAsyncConfiguration配置類
3.1 ProxyAsyncConfiguration
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
@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(); 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; }} |
作用也很明顯，就是往spring容器中添加了AsyncAnnotationBeanPostProcessor類
3.2 AsyncAnnotationBeanPostProcessor
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
public class AsyncAnnotationBeanPostProcessor extends AbstractBeanFactoryAwareAdvisingPostProcessor { // 刪除了一些無關緊要，或者默認不會設置的屬性 public AsyncAnnotationBeanPostProcessor() { setBeforeExistingAdvisors(true); } /** * 因為AsyncAnnotationBeanPostProcessor實現(xiàn)了BeanFactoryAware接口 * 所以在實例化的過程中執(zhí)行到initializeBean步驟的時候，里面第一步就是執(zhí)行各種實現(xiàn)了Aware接口的接口方法 * 在此處new了一個advisor。advisor簡單理解就是：advice+pointcut * @param beanFactory */ @Override public void setBeanFactory(BeanFactory beanFactory) { super.setBeanFactory(beanFactory); AsyncAnnotationAdvisor advisor = new AsyncAnnotationAdvisor(this.executor, this.exceptionHandler); if (this.asyncAnnotationType != null) { advisor.setAsyncAnnotationType(this.asyncAnnotationType); } advisor.setBeanFactory(beanFactory); this.advisor = advisor; }} |
其實可以看到最重要的方法，就是setBeanFactory了，該方法是在AsyncAnnotationBeanPostProcessor的生命周期最后一步initializeBean里面的第一小步，也就是執(zhí)行所有Aware接口的時候執(zhí)行。
對于AOP來說，其實最主要的就是advice+pointcut，也就是advisor，在生命周期的這一步，也創(chuàng)建了advisor。
3.3 AsyncAnnotationAdvisor
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
public AsyncAnnotationAdvisor( @Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) { Set<Class<? extends Annotation>> asyncAnnotationTypes = new LinkedHashSet<>(2); /** * 這兒設置符合pointCut需要的注解 * 此處的executor就是一個擴展點，如果不想用spring的默認單線程線程池，可以自定義一個線程池 * exceptionHandler，顧名思義，就是我們的方法在線程池中執(zhí)行時拋出exception該如何handle使用的 * advice也就是咱們的interceptor * pointCut就不多解釋了，就是把設置符合什么條件會進行interceptor的invoke方法 */ asyncAnnotationTypes.add(Async.class); try { 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. } this.advice = buildAdvice(executor, exceptionHandler); this.pointcut = buildPointcut(asyncAnnotationTypes); } |
可以看到最主要的工作就是buildAdvice和buildPointcut。advice的作用是定義在方法執(zhí)行方面，該如何執(zhí)行；pointcut的作用是定義方法的范圍
3.3.1 buildAdvice
|
1
2
3
4
5
6
7
|
protected Advice buildAdvice( @Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) { // new了一個interceptor AnnotationAsyncExecutionInterceptor interceptor = new AnnotationAsyncExecutionInterceptor(null); interceptor.configure(executor, exceptionHandler); return interceptor; } |
可以看到advice主要就是定義了一個爛機器interceptor，在方法執(zhí)行的時候進行一些攔截，至于executor，是方法執(zhí)行器，默認為null，exceptionHandler也默認是null。
3.3.1.1 AnnotationAsyncExecutionInterceptor，異步執(zhí)行的原理
在AnnotationAsyncExecutionInterceptor的父類AsyncExecutionInterceptor中，實現(xiàn)了攔截器的接口方法invoke，也就是真實的方法執(zhí)行邏輯。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
|
/** * Intercept the given method invocation, submit the actual calling of the method to * the correct task executor and return immediately to the caller. * @param invocation the method to intercept and make asynchronous * @return {@link Future} if the original method returns {@code Future}; {@code null} * otherwise. */ @Override @Nullable 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í)行器 * 1. 從@Async注解里面獲取配置的任務執(zhí)行器 * 2. 從Spring容器中找TaskExecutor類的bean * 3. 從spring容器中獲取名為"taskExecutor"的bean， * 4. 如果還沒有，new SimpleAsyncTaskExecutor()） */ AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod); if (executor == null) { throw new IllegalStateException( "No executor specified and no default executor set on AsyncExecutionInterceptor either"); } //將當前方法執(zhí)行封裝成一個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()); } |
可以看到主要做的事情是:
- 尋找任務執(zhí)行器：
- 從@Async注解里面獲取配置的任務執(zhí)行器
- 從Spring容器中找TaskExecutor類的bean
- 從spring容器中獲取名為"taskExecutor"的bean，
- 如果還沒有，new SimpleAsyncTaskExecutor()）可以看到其實我們是可以給@Async進行任務執(zhí)行器的配置的。
- 將具體的方法封裝成callable的對象，然后doSubmit
- 此處我們就看一下默認的doSumit，使用的SimpleAsyncTaskExecutor是如何實現(xiàn)的
- 最終會執(zhí)行到下面這個doExecute方法，默認情況下threadFactory是null，所以默認情況下，我們的方法，每次都是被創(chuàng)建了一個新的守護線程來進行方法的執(zhí)行。
|
1
2
3
4
|
protected void doExecute(Runnable task) { Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task)); thread.start();} |
3.3.1.2 自定義任務執(zhí)行器
- 可以在配置類里new SimpleAsyncTaskExecutor()，然后setThreadFactory，這樣修改了默認線程的產生方式
- 比較主流的方式是，定義一個ThreadPoolTaskExecutor，也就是線程池任務執(zhí)行器，可以進行線程復用
3.3.2 buildPointcut
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
/** * Calculate a pointcut for the given async annotation types, if any. * @param asyncAnnotationTypes the async annotation types to introspect * @return the applicable Pointcut object, or {@code null} if none */ protected Pointcut buildPointcut(Set<Class<? extends Annotation>> asyncAnnotationTypes) { ComposablePointcut result = null; for (Class<? extends Annotation> asyncAnnotationType : asyncAnnotationTypes) { // 就是根據(jù)這兩個匹配器進行匹配的 // 檢查類上是否有@Async注解 Pointcut cpc = new AnnotationMatchingPointcut(asyncAnnotationType, true); //檢查方法上是否有@Async注解 Pointcut mpc = new AnnotationMatchingPointcut(null, asyncAnnotationType, true); if (result == null) { result = new ComposablePointcut(cpc); } else { // 取并集：類上加了@Async或者類的方法上加了@Async result.union(cpc); } result = result.union(mpc); } return (result != null ? result : Pointcut.TRUE); } |
主要方法就是定義了一個類匹配pointcut和一個方法匹配pointcut。
4 什么時候判斷進行advice的添加呢
當然就是在對某個bean進行proxy的判斷的時候，也就是bean的生命周期最后一步，也是initializeBean里最后的一步，對于BeanPostProcessor的執(zhí)行
3.4.1 AsyncAnnotationBeanPostProcessor#postProcessAfterInitialization
要注意的是AsyncAnnotationBeanPostProcessor的postProcessAfterInitialization方法其實是繼承的是父類AbstractAdvisingBeanPostProcessor的。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
@Override public Object postProcessAfterInitialization(Object bean, String beanName) { // 沒有通知，或者是AOP的基礎設施類，那么不進行代理 if (this.advisor == null || bean instanceof AopInfrastructureBean) { // Ignore AOP infrastructure such as scoped proxies. return bean; } // 對已經被代理的類，不再生成代理，只是將通知添加到代理類的邏輯中 // 這里通過beforeExistingAdvisors決定是將通知添加到所有通知之前還是添加到所有通知之后 // 在使用@Async注解的時候，beforeExistingAdvisors被設置成了true， // @Async注解之所以把beforeExistingAdvisors設置為true，是因為該advisor和其他的advisor差別太大了，從情理上講，也應該第一個執(zhí)行 // 意味著整個方法及其攔截邏輯都會異步執(zhí)行 if (bean instanceof Advised) { Advised advised = (Advised) bean; // 判斷bean是否符合該advisor的使用范圍，通過pointcut來判斷 if (!advised.isFrozen() && isEligible(AopUtils.getTargetClass(bean))) { // Add our local Advisor to the existing proxy's Advisor chain... if (this.beforeExistingAdvisors) { advised.addAdvisor(0, this.advisor); } else { advised.addAdvisor(this.advisor); } return bean; } } // 如果還不是一個代理類，也需要通過eligible來判斷是否符合使用該advisor的條件 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()); } // No proxy needed. return bean; } |
而在isEligible中，就是判斷當前執(zhí)行生命周期的bean是否滿足我們的@Async注解的使用范圍，主要是通過其class來判斷
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
protected boolean isEligible(Class<?> targetClass) { Boolean eligible = this.eligibleBeans.get(targetClass); if (eligible != null) { return eligible; } if (this.advisor == null) { return false; } // 其實就是判斷類是否可以進行添加該advisor，也就是判斷是否符合該advisor的使用條件 // 就是把advisor的pointCut拿出來，pointCut里的classMatcher和methodMatcher拿出來對類及其方法進行判斷 eligible = AopUtils.canApply(this.advisor, targetClass); this.eligibleBeans.put(targetClass, eligible); return eligible; } |
具體的AopUtils.canApply(this.advisor, targetClass)邏輯就不寫了，就是根據(jù)pointcut里設置的classFilter和methodMatcher類判斷當前bean的class是否需要進行該advisor的使用。
總結
發(fā)現(xiàn)@Async注解還是挺麻煩的，特別是要寫一篇簡單易懂的博客，更難。
默認配置實現(xiàn)原理：在執(zhí)行的時候將method最終封裝成一個Runable對象，然后new一個線程，通過線程的start方法，進行method的執(zhí)行，來實現(xiàn)異步。
到此這篇關于Java Spring之@Async原理案例詳解的文章就介紹到這了,更多相關Java Spring之@Async原理內容請搜索服務器之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持服務器之家！
原文鏈接：https://blog.csdn.net/liangsheng_g/article/details/119852868
