spring异步线程池源码解读

• 基本使用
• • 配置异步线程池，开始异步线程
  • 使用
• 源码解读
• • 开启后怎么生效的
  • • @EnableAsync
    • AsyncConfigurationSelector
    • ProxyAsyncConfiguration
    • AsyncAnnotationBeanPostProcessor
    • 请求进入，创建代理
  • 怎么异步执行的
  • • AsyncExecutionInterceptor
    • AsyncExecutionAspectSupport
  • 其他
  • • 判断需要创建代理
    • • filter定义的地方
• 总结

基本使用

配置异步线程池，开始异步线程

// 开始异步线程池使用
@EnableAsync
@Configuration
public class AsyncThreadConfig {@Resourceprivate AsyncThreadConfigProperties properties;// 配置异步线程池@Bean("async-1")@Lazypublic ThreadPoolTaskExecutor threadPoolAsync1(){ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(properties.getCorePoolSize1());executor.setMaxPoolSize(properties.getMaxPoolSize1());executor.setKeepAliveSeconds(properties.getKeepAliveSeconds1());executor.setQueueCapacity(properties.getQueueCapacity1());executor.setThreadNamePrefix(properties.getThreadNamePrefix1());executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());executor.setWaitForTasksToCompleteOnShutdown(true);executor.setAwaitTerminationSeconds(properties.getAwaitTerminationSeconds1());executor.initialize();return executor;}
}

使用

@Service
public class AsyncService {private static Logger logger= LoggerFactory.getLogger(AsyncService.class);@Async("async-1")public void printTime(){SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");logger.info("调用时间 {}",sdf.format(new Date()));}
}

源码解读

开启后怎么生效的

@EnableAsync

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
// 可以看到这里导入了一个  AsyncConfigurationSelector
@Import(AsyncConfigurationSelector.class)
public @interface EnableAsync {
...
}

AsyncConfigurationSelector

// 会根据@EnableAsync中的mode字段导入不同的配置类，
// 这里默认为PROXY，即使用ProxyAsyncConfiguration配置类
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;}
}

ProxyAsyncConfiguration

// 这里创建了一个 AsyncAnnotationBeanPostProcessor实例加入spring容器
@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 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.getNumber("order"));return bpp;
}

AsyncAnnotationBeanPostProcessor

// 这里创建了一个AsyncAnnotationAdvisor类，最终会构造出一个AnnotationAsyncExecutionInterceptor
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;
}// AsyncAnnotationAdvisor 构造方法
public AsyncAnnotationAdvisor(@Nullable Supplier executor, @Nullable Supplier exceptionHandler) {Set> asyncAnnotationTypes = new LinkedHashSet<>(2);asyncAnnotationTypes.add(Async.class);try {asyncAnnotationTypes.add((Class)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);
}protected Advice buildAdvice(@Nullable Supplier executor, @Nullable Supplier exceptionHandler) {// AsyncAnnotationAdvisor#buildAdviceAnnotationAsyncExecutionInterceptor interceptor = new AnnotationAsyncExecutionInterceptor(null);interceptor.configure(executor, exceptionHandler);return interceptor;
}

请求进入，创建代理

// AbstractAdvisingBeanPostProcessor#postProcessAfterInitialization 
// 这里的实例是AsyncAnnotationBeanPostProcessor，也就是上面创建并加入的实例类
// AbstractAdvisingBeanPostProcessor#postProcessAfterInitialization
public Object postProcessAfterInitialization(Object bean, String beanName) {// ... 部分代码省略// 判断是否符合代理条件// 对类或者方法进行校验(判断类或者方法有没有对应的注解注释)if (isEligible(bean, beanName)) {ProxyFactory proxyFactory = prepareProxyFactory(bean, beanName);if (!proxyFactory.isProxyTargetClass()) {evaluateProxyInterfaces(bean.getClass(), proxyFactory);}proxyFactory.addAdvisor(this.advisor);customizeProxyFactory(proxyFactory);// Use original ClassLoader if bean class not locally loaded in overriding class loaderClassLoader classLoader = getProxyClassLoader();if (classLoader instanceof SmartClassLoader && classLoader != bean.getClass().getClassLoader()) {classLoader = ((SmartClassLoader) classLoader).getOriginalClassLoader();}return proxyFactory.getProxy(classLoader);}// No proxy needed.return bean;
}

怎么异步执行的

AsyncExecutionInterceptor

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);// 获取 executor AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod);if (executor == null) {throw new IllegalStateException("No executor specified and no default executor set on AsyncExecutionInterceptor either");}Callable