spring context源码解析之@Async

1、背景

本文转自“天河聊技术”微信公众号

你们项目中用到异步、多线程的场景不少，使用最多的场景仍是主动对象模式，就是主线程开启一个线程池去任务分发，任务执行完成以后，关闭线程池，可是有的场景则须要部分代码异步执行的效果，简单的说就是有一个能够复用的线程池能够复用，直接new Thread固然也能够，不推荐，没有线程池的可靠性好，若是这个时候再建立一个线程池用完再关闭代码是否是有点重，维护性也很差，@Async这个注解就是为了解决这个问题，只须要在bean的方法上加上这和注解便可，就能够实现异步。

 

2、正文

先简单介绍下怎么使用

一、先在程序上加上@EnableAsync这个注解，能够指定执行器，若是不指定就从beanFactory中获取taskExecutor这个执行器

@Async(value = "executor")
    public void test(){

二、若是想自定义执行器，实现这个接口org.springframework.scheduling.annotation.AsyncConfigurer或者继承org.springframework.scheduling.annotation.AsyncConfigurerSupport这个类

public class AsyncExecutor implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
//        自定义线程池
        return Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    }

 

源码实现

@EnableAsync这个注解作什么了什么事情

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(AsyncConfigurationSelector.class)
public @interface EnableAsync {

能够看到@Import引入了AsyncConfigurationSelector这个类，前面介绍spring ioc源码解析的时候这个@Import这个注解会把引入的类进行bean定义解析后注册到beanFactory然后进行实例化。

 

咱们看到这个selector又继承了这个类

public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> {

spring切面拦截到这个注解后进行下一步操做。

 

org.springframework.scheduling.annotation.AsyncConfigurationSelector#selectImports

有两种通知模式实现。

 

咱们能够看到@EnableAsync这个注解中是能够指定mode属性的，若是不指定默认设置是PROXY

AdviceMode mode() default AdviceMode.PROXY;

咱们这里有以PROXY为例跟踪下。

 

会加载到这个类

org.springframework.scheduling.annotation.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，BeanPostProcessor 前面spring ioc源码解析部分介绍过是对bean初始化过程的管理
      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);
      }
//    这里默认是jdk的动态代理，能够@EnableAsync这个注解的属性值，也能够指定
      bpp.setProxyTargetClass(this.enableAsync.getBoolean("proxyTargetClass"));
      bpp.setOrder(this.enableAsync.<Integer>getNumber("order"));
      return bpp;
   }

}

进入

org.springframework.scheduling.annotation.AbstractAsyncConfiguration

@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();
   }

这个方法对线程进行赋值。

咱们能够看到这个接口org.springframework.scheduling.annotation.AsyncConfigurer的惟一实现类org.springframework.scheduling.annotation.AsyncConfigurerSupport

public class AsyncConfigurerSupport implements AsyncConfigurer {

   @Override
   public Executor getAsyncExecutor() {
      return null;
   }

   @Override
   @Nullable
   public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
      return null;
   }

}

是一个空实现，多以想要自定义本身的线程池，实现这个接口org.springframework.scheduling.annotation.AsyncConfigurer或者继承org.springframework.scheduling.annotation.AsyncConfigurerSupport重写父类的方法。

 

异步方法返回值支持void、java.util.concurrent.Future、org.springframework.util.concurrent.ListenableFuture、java.util.concurrent.CompletableFuture

 

进入

org.springframework.aop.interceptor.AsyncExecutionInterceptor

找到

org.springframework.aop.interceptor.AsyncExecutionInterceptor#invoke这个方法，执行目标方法的切面方法

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

      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());
   }

进入

org.springframework.aop.interceptor.AsyncExecutionAspectSupport#determineAsyncExecutor

@Nullable
   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;
            if (targetExecutor == null) {
               synchronized (this.executors) {
                  if (this.defaultExecutor == null) {
//                   从beanFactory中获取taskExecutor执行器
                     this.defaultExecutor = getDefaultExecutor(this.beanFactory);
                  }
                  targetExecutor = this.defaultExecutor;
               }
            }
         }
         if (targetExecutor == null) {
            return null;
         }
         executor = (targetExecutor instanceof AsyncListenableTaskExecutor ?
               (AsyncListenableTaskExecutor) targetExecutor : new TaskExecutorAdapter(targetExecutor));
//       放入缓存
         this.executors.put(method, executor);
      }
      return executor;
   }

若是@Async指定了执行器，就会使用这个执行器，若是没有指定就从beanFactory中获取taskExecutor这个执行器。

 

 

3、最后

本次源码解析到这里，仅供参考。