@Async 정리

스프링에서 특정 메서드를 비동기적으로 수행하기 위해서 @Async를 사용한다.

 

@Async 메서드가 포함된 빈 등록

@Async가 메서드에 선언된 빈은 프록시 객체로 등록된다.

 

1. @EnableAsync로 AsyncConfigurationSelector가 임포트 된다.

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

 

2. 별도의 설정을 하지 않으면 AdviceMode의 기본 값은 PROXY로 AsyncConfigurationSelector에서 ProxyAsyncConfiguration가 선택 되어 AsyncAnnotationBeanPostProcessor가 빈으로 등록된다.

public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> {  

  public String[] selectImports(AdviceMode adviceMode) {  
       return switch (adviceMode) {  
          case PROXY -> new String[] {ProxyAsyncConfiguration.class.getName()};  
          case ASPECTJ -> new String[] {ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME};  
       };  
    }  

}

 

3. AsyncAnnotationBeanPostProcessor가 생성 되며 @Async가 선언된 클래스를 지정하는 AsyncAnnotationAdvisor가 추가된다.

@Configuration(proxyBeanMethods = false)  
@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();  
       // executor와 exceptionHandler를 설정한다.
       bpp.configure(this.executor, this.exceptionHandler);  
       // EnableAsync가 아닌 커스텀 어노테이션을 사용한다면 해당 어노테이션을 사용할 수 있도록 처리한다.
       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는 BeanFactoryAware의 setBeanFactory 메서드의 구현하고 있다.
setBeanFactory에서 BeanFactory를 주입받아 AbstractAdvisingBeanPostProcessor에 필요한 AsyncAnnotationAdvisor를 생성한다.

 

4. BeanPostProcessor를 구현한 AbstractAdvisingBeanPostProcessor의 postProcessAfterInitialization 메서드를 통해 프록시 객체가 생성, 등록된다.

 

@Async 메서드가 포함된 빈 실행

AsyncExecutionInterceptor에 의한 실행 가로채기

@Async가 선언된 메서드는 실행시 AsyncExecutionInterceptor에 의해 실행이 가로채진다.

public class AsyncExecutionInterceptor extends AsyncExecutionAspectSupport implements MethodInterceptor, Ordered {
    // ...

    @Override  
    @Nullable  
    public Object invoke(final MethodInvocation invocation) throws Throwable {
        // ...
    }
}

 

Executor 설정

// AsyncExecutionInterceptor
AsyncTaskExecutor executor = determineAsyncExecutor(userMethod);

// AsyncExecutionAspectSupport
@Nullable  
protected AsyncTaskExecutor determineAsyncExecutor(Method method) {  
    // executors는 executor 캐싱을 위한 것  
    AsyncTaskExecutor executor = this.executors.get(method);  
    if (executor == null) {  
        Executor targetExecutor;  
        // @Async 메서드에 설정된 값   
String qualifier = getExecutorQualifier(method);  
        if (this.embeddedValueResolver != null && StringUtils.hasLength(qualifier)) {  
            qualifier = this.embeddedValueResolver.resolveStringValue(qualifier);  
        }  
        if (StringUtils.hasLength(qualifier)) {  
            // qualifier에 해당하는 Executor를 찾는다.  
            targetExecutor = findQualifiedExecutor(this.beanFactory, qualifier);  
        }  
        else {  
            // 기본 Executor를 사용한다.  
            targetExecutor = this.defaultExecutor.get();  
        }  
        if (targetExecutor == null) {  
            return null;  
        }  
        executor = (targetExecutor instanceof AsyncTaskExecutor asyncTaskExecutor ?  
                asyncTaskExecutor : new TaskExecutorAdapter(targetExecutor));  
        // executor를 method를 기준으로 캐싱한다.  
        this.executors.put(method, executor);  
    }  
    return executor;  
}

가로챈 메서드를 실행하기 전 가장 먼저 어떤 Executor에서 해당 메서드를 실행할 것인지 정한다.
@Async에 값이 사용할 Executor 이름이 지정되어 있다면 해당하는 Executor를 그렇지 않다면 기본 Executor를 사용한다.

 

Task 정의 및 제출

Callable<Object> task = () -> {  
    try {  
       Object result = invocation.proceed();  
       if (result instanceof Future<?> future) {  
          return future.get();  
       }  
    }  
    catch (ExecutionException ex) {  
       handleError(ex.getCause(), userMethod, invocation.getArguments());  
    }  
    catch (Throwable ex) {  
       handleError(ex, userMethod, invocation.getArguments());  
    }  
    return null;  
};  

return doSubmit(task, executor, invocation.getMethod().getReturnType());

Executor가 정해지면 Task를 정의하고 제출한다.

 

Task 제출 반환 값

@Nullable  
@SuppressWarnings("deprecation")  
protected Object doSubmit(Callable<Object> task, AsyncTaskExecutor executor, Class<?> returnType) {  
    if (CompletableFuture.class.isAssignableFrom(returnType)) {  
       return executor.submitCompletable(task);  
    }  
    else if (org.springframework.util.concurrent.ListenableFuture.class.isAssignableFrom(returnType)) {  
       return ((org.springframework.core.task.AsyncListenableTaskExecutor) executor).submitListenable(task);  
    }  
    else if (Future.class.isAssignableFrom(returnType)) {  
       return executor.submit(task);  
    }  
    else if (void.class == returnType || "kotlin.Unit".equals(returnType.getName())) {  
       executor.submit(task);  
       return null;  
    }  
    else {  
       throw new IllegalArgumentException(  
             "Invalid return type for async method (only Future and void supported): " + returnType);  
    }  
}

doSubmit의 구현을 확인해 보면 AsyncTaskExecutor로 처리할 수 있는 반환 타입은 아래와 같다는 것을 알 수 있다.

• CompletableFuture
• ListenableFuture
• Future
• void