Spring 비동기 처리 @Async, CompletableFuture, TaskDecorator

`비동기 처리`는 작업을 별도의 스레드에서 실행하고 결과를 나중에 처리하는 방식입니다. 이를 통해 특정 로직의 실행이 끝날 때 까지 기다리지 않고 다음 코드를 실행할 수 있으며, 결과가 준비되면 이벤트를 받거나 콜백을 통해 처리합니다. 

---

Spring Framework 비동기 처리 방식

• `Callable`과 `DeferredResult`
  • Spring Framework에서 스레드 기반의 비동기 처리를 지원합니다.
  • `Callable`: 비동기 작업을 `Callable`로 감싸고 `AsyncTaskExecutor`를 통해 별도의 스레드에서 실행합니다. 작업이 완료되면 `Callable`을 반환합니다.
  • `DeferredResult`: 비동기 작업의 결과를 나중에 처리할 수 있는 `DeferredResult`를 반환합니다. 작업이 완료된 후 반환한 `DeferredResult`에 결과를 세팅하고 응답을 보낼 수 있습니다.
    
    
• `@Async` ⭐ ⭐ ⭐
  @Async 어노테이션은 메서드에 적용하여 해당 메서드를 비동기적으로 실행하게 하는 방식입니다. 이 방식은 비교적 간단하게 비동기 처리를 구현할 수 있으며, 비동기 작업의 결과를 `Future`나 `CompletableFuture`로 반환할 수 있습니다. 또한, Spring이 제공하는 `TaskExecutor` 빈을 통해 스레드 풀을 구성하고 작업을 분산 처리할 수도 있습니다.
  
  
• ListenableFuture
  ListenableFuture 인터페이스는 비동기 작업의 결과에 대한 콜백 메서드를 등록할 수 있도록 합니다. 콜백 메서드를 통해 논블로킹 처리가 가능합니다. 첫번째 파라미터는 성공시 실행할 것을, 두 번째 파라미터는 실패시 실행할 것을 지정해주면 됩니다.
  
  
• CompletableFuture
  CompletableFuture는 Java 8에서 추가된 클래스로, 비동기 작업의 결과를 처리하기 위한 기능을 제공합니다.  ListenableFuture와 유사한 기능을 제공하며, 추가적으로 비동기 작업의 연결, 조합, 변환 등을 쉽게 처리할 수 있습니다.

 

---

@Async 사용을 위한 비동기 기능 설정

1. @EnableAsync 활성화

설정 클래스에 `@EnableAsync`를 적용하여, Spring Framework가 비동기 처리를 지원하도록 활성화합니다.

@Configuration
@EnableAsync
public class AppConfig {
    // 설정 내용
}

2. TaskExecutor 빈 설정

비동기 작업을 처리할 `스레드 풀`을 구성하기 위해 `TaskExecutor` 빈을 설정해줍니다.

아래 예제 코드에서는 Spring이 제공하는 구현체 중 하나인 `ThreadPoolTaskExecutor`를 사용하였습니다. 또한 필요에 따라 세밀한 제어를 위해 스레드 풀의 크기 설정을 조정할 수 있습니다.

아래 예제는 `getAsyncExecutor()`메서드를 오버라이드하여 `ThreadPoolTaskExecutor`를 구성하여 반환한 예제입니다.

@Configuration
@EnableAsync
public class AppConfig implements AsyncConfigurer {

    private int CORE_POOL_SIZE = 10;
    private int MAX_POOL_SIZE = 100;
    private int QUEUE_CAPACITY = 1000;

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(CORE_POOL_SIZE);  // 스레드 풀의 기본 크기
        executor.setMaxPoolSize(MAX_POOL_SIZE);    // 스레드 풀의 최대 크기
        executor.setQueueCapacity(QUEUE_CAPACITY); // 대기열 크기
        executor.initialize();                     // 스레드 풀 초기화
        return executor;
    }
}

3. @Async 어노테이션 적용

비동기로 실행할 메서드에 `@Async`를 부여해주면 비동기 처리(별도의 쓰레드)를 수행합니다. 호출자는 이 메서드를 호출하고 나서 다른 작업을 수행할 수 있습니다.

비동기 메서드를 호출하면Spring은 내부적으로 `TaskExecutor`를 사용하여 비동기 작업을 처리하고 작업이 완료되면 `Future` 객체나 `CompletableFuture`를 반환하여 비동기 작업의 결과를 처리할 수 있습니다.

@Service
public class AsyncService {

    @Async
    public void asyncMethod() {
        // 비동기로 실행될 코드
    }
}

4. 호출자에서 비동기 메서드 호출

비동기 메서드인 `asyncMethod()`를 호출하는 예제입니다. 이때 호출자는 `asyncMethod()`가 실행 중일 때 다른 작업을 수행할 수 있습니다.

@Controller
public class AsyncController {
    
    @Autowired
    private AsyncService asyncService;
    
    @GetMapping("/doAsync")
    public String doAsync() {
    
        // 비동기 메서드 호출
        myService.asyncMethod();
        
        // ... 다른 작업들
        // (호출자는 비동기 메서드가 실행 중일 때 다른 작업 수행 가능)
        
        return "result";
    }
}

 

---

@Async과 CompletableFuture

`@Async` 어노테이션은 메서드에 적용되며, 해당 메서드를 비동기적으로 실행하도록 지정합니다. 메서드가 호출되면 Spring은 내부적으로 별도의 스레드를 생성하여 작업을 비동기적으로 실행합니다. `@Async` 어노테이션은 메서드의 반환 타입이 `void` 또는 `Future<T>`일 수 있습니다. 반환 타입이 `Future<T>`인 경우, 비동기 작업이 완료되면 `Future` 객체를 통해 결과를 얻을 수 있습니다.

`CompletableFuture`는 Java 8에서 추가된 클래스로, 비동기 작업의 결과를 처리하기 위한 기능을 제공합니다. `CompletableFuture`는 `Future` 인터페이스를 확장하고, 비동기 작업의 결과를 처리하거나 다른 비동기 작업을 조합하는데 사용됩니다. `CompletableFuture`는 다양한 메서드를 제공하여 작업을 구성하고 조합할 수 있으며, 비동기 작업의 결과를 처리하기 위한 콜백 메서드를 등록할 수 있습니다.

`@Async` 어노테이션을 사용하면 Spring은 내부적으로 비동기 작업을 처리하기 위해 `CompletableFuture`를 활용합니다. 예를 들어, `@Async` 어노테이션이 적용된 메서드는 `CompletableFuture.supplyAsync` 또는 `CompletableFuture.runAsync` 메서드를 호출하여 해당 메서드를 비동기적으로 실행합니다. `CompletableFuture`는 비동기 작업의 결과를 `Future` 객체로 래핑하여 반환하므로, 비동기 작업이 완료된 후에 결과를 처리할 수 있습니다.

 

즉, `@Async` 어노테이션을 사용하여 메서드를 비동기적으로 실행하고, `CompletableFuture`를 활용하여 비동기 작업의 결과를 처리할 수 있습니다.

 

예시1: 기본 동작

1. 비동기 작업을 수행할 메서드에 `@Async` 어노테이션을 적용하고 메서드는 `CompletableFuture`를 반환하도록 설정합니다.

@Service
public class AsyncService {
    
    @Async
    public CompletableFuture<String> doAsyncTask() {
        // 비동기적으로 실행되어야 하는 작업을 수행합니다.
        // 작업이 완료되면 결과를 CompletableFuture에 설정합니다.
        String result = "비동기 작업 완료";
        return CompletableFuture.completedFuture(result);
    }
}

 

2. 비동기 작업을 호출한 후, CompletableFuture를 사용하여 결과를 처리합니다.

@RestController
public class AsyncController {
    
    private final AsyncService asyncService;
    
    public MyController(AsyncService asyncService) {
        this.asyncService = asyncService;
    }
    
    @GetMapping("/async-task")
    public CompletableFuture<String> handleAsyncTask() {
    
        // 비동기 작업을 호출하고 CompletableFuture를 반환받습니다.
        CompletableFuture<String> futureResult = asyncService.doAsyncTask();
        
        // CompletableFuture를 통해 비동기 작업의 결과를 처리합니다.
        futureResult.thenApply(result -> {
            // 비동기 작업이 완료된 후 실행될 로직을 작성합니다.
            // result는 비동기 작업의 결과입니다.
            return "비동기 작업 결과: " + result;
        });
        
        return futureResult;
    }
}

`CompletableFuture` 메서드

`CompletableFuture`를 사용하면 비동기 작업의 결과를 콜백 함수나 연산 체인을 통해 처리할 수 있습니다. 아래는 `CompletableFuture`의 메서드 중 일부입니다.

• thenApply(Function<? super T,? extends U> fn)
  • CompletableFuture의 결과를 가공하고 변환하는데 사용됩니다.
  • Function<T, U>를 인자로 받아 결과를 변환한 CompletableFuture<U>를 반환합니다.
• thenAccept(Consumer<? super T> action)
  
  • CompletableFuture의 결과를 소비하고 추가 작업을 수행하는데 사용됩니다.
  • Consumer<T>를 인자로 받아 추가 작업을 수행합니다.
• thenRun(Runnable action)
  
  • CompletableFuture의 결과를 사용하지 않고 추가 작업을 수행하는데 사용됩니다.
  • Runnable을 인자로 받아 추가 작업을 수행합니다.
• exceptionally(Function<Throwable,? extends T> fn)
  
  • CompletableFuture에서 발생한 `예외`를 처리하는데 사용됩니다.
  • Function<Throwable, T>를 인자로 받아 예외를 처리하고 대체 값을 반환하는 CompletableFuture<T>를 반환합니다.
• handle(BiFunction<? super T,Throwable,? extends U> fn)
  
  • CompletableFuture의 결과 또는 `예외`를 처리하는데 사용됩니다.
  • BiFunction<T, Throwable, U>를 인자로 받아 결과 또는 예외를 처리하고 결과를 반환하는 CompletableFuture<U>를 반환합니다.
• whenComplete(BiConsumer<? super T,? super Throwable> action)
  
  • CompletableFuture의 결과 또는 `예외`를 소비하고 추가 작업을 수행하는데 사용됩니다.
  • BiConsumer<T, Throwable>를 인자로 받아 결과 또는 예외를 처리하고 추가 작업을 수행합니다.
• allOf(CompletableFuture<?>... cfs)
  
  • 여러 개의 CompletableFuture를 동시에 실행하고, 모든 CompletableFuture의 완료를 기다리는데 사용됩니다.
  • 모든 CompletableFuture가 완료되면 CompletableFuture<Void>를 반환합니다.
• anyOf(CompletableFuture<?>... cfs)
  
  • 여러 개의 CompletableFuture 중 가장 빨리 완료된 CompletableFuture를 반환하는데 사용됩니다.
  • 가장 빨리 완료된 CompletableFuture의 결과 타입을 반환합니다.

 

예시2: 예외 발생

@Service
public class AsyncService {

    @Async
    public CompletableFuture<String> doSomethingAsync() {
        try {
            // 비동기로 수행할 작업
            // ...
            // 예외 발생 가능성이 있는 작업
            if (someCondition) {
                throw new RuntimeException("오류!");
            }
            return CompletableFuture.completedFuture("작업 성공!");
        } catch (Exception e) {
            return CompletableFuture.failedFuture(e);
        }
    }
}

• 작업이 성공하면 결과 값을 감싸고, 예외가 발생하면 예외를 감싼 `CompletableFuture`를 반환합니다.
• `failedFuture()` 메서드를 사용하여 예외를 감싸지 않고 예외를 바로 던지더라도 상관없습니다.  
  `filedFuture()`를 사용하는 이유는 예외를 명시적으로 처리할 수 있기 때문입니다.

public class MyApp {

    public static void main(String[] args) {
        AsyncService asyncService = new AsyncService();

        CompletableFuture<String> future = asyncService.asyncMethod();

        future.thenAccept(result -> {
            // 작업이 완료되었을 때 결과 처리
            System.out.println("결과: " + result);
        }).exceptionally(e -> {
            // 예외 처리
            System.err.println("예외 발생: " + e.getMessage());
            return null;
        });
    }
}

`CompletableFuture`에서 예외가 발생한 경우에 `exceptionally()`가 호출됩니다. 이를 통해 예외가 발생하면 해당 예외를 처리하는 로직을 수행할 수 있습니다.

---

TaskDecorator: 비동기 작업 데코레이터

`TaskDecorator`는 Spring의 TaskExecutor에 대한 데코레이터 인터페이스입니다.

비동기 작업을 실행하는 동안 작업 실행 전후에 추가적인 작업을 수행할 수 있도록 해줍니다.

이를 통해 비동기 작업의 작업 실행 컨텍스트를 변경하거나, 작업에 대한 추가적인 로깅, 보안 체크, 성능 모니터링 등을 수행할 수 있습니다.

import org.springframework.core.task.TaskDecorator;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;

public class CustomTaskDecorator implements TaskDecorator {

    @Override
    public Runnable decorate(Runnable runnable) {
        // 현재 요청의 RequestAttributes를 가져옴
        RequestAttributes attributes = RequestContextHolder.currentRequestAttributes();
        
        return () -> {
            try {
                // 작업 실행 전에 RequestAttributes를 설정
                RequestContextHolder.setRequestAttributes(attributes);
                
                // 작업 실행
                runnable.run();
            } finally {
                // 작업 실행 후에 RequestAttributes를 제거
                RequestContextHolder.resetRequestAttributes();
            }
        };
    }
}

• `decorate()`메서드를 오버라이드해서 작업 실행 전후에 추가적인 작업을 수행할 수 있도록 하였습니다.

@Configuration
@EnableAsync
public class AsyncConfig {

    private int CORE_POOL_SIZE = 10;
    private int MAX_POOL_SIZE = 100;
    private int QUEUE_CAPACITY = 1000;

    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(CORE_POOL_SIZE);  // 스레드 풀의 기본 크기
        executor.setMaxPoolSize(MAX_POOL_SIZE);    // 스레드 풀의 최대 크기
        executor.setQueueCapacity(QUEUE_CAPACITY); // 대기열 크기
        executor.setThreadNamePrefix("async-");    // 스레드 접두어
        
        // CustomTaskDecorator를 TaskExecutor에 등록
        executor.setTaskDecorator(new CustomTaskDecorator());
        
        executor.initialize();
        return executor;
    }
}

`AsyncConfigurer`를 구현하는 방식과 `Executor`를 빈으로 등록하는 방식은 같습니다. 차이점은 복잡하고 많은 커스터마이징이 필요한 경우 `AsyncConfigurer`를 구현하는 방식이 용이하다는 점입니다.

 

---

RejectedExecutionHandler: 거부 작업 처리

스레드 풀이 작업을 처리할 수 있는 한계에 도달하면, 새로운 요청 작업이 거부될 수 있습니다. 이런 경우 `RejectetdExecutionHandler`를 사용하여 처리 방식을 지정할 수 있습니다.

ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

//...

executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();

• `AbortPolicy`: Default, 거부된 실행 요청이 발생하면 `RejectedExecutionException`을 던지고 예외를 발생시킵니다.
• `CallerRunsPolicy`: 거부된 실행 요청이 발생하면 해당 요청을 호출한 스레드에서 직접 실행합니다. 이 경우, 스레드 풀이 작업을 처리할 수 있는 능력을 초과하는 경우에도 실행 요청을 처리할 수 있는 임시 방법으로 사용될 수 있습니다. 하지만 호출 스레드는 블로킹될 수 있으므로 주의해야 합니다.
• `DiscardPolicy`: 거부된 실행 요청을 무시합니다.
• `DiscardOldestPolicy`: 거부된 실행 요청 대신 가장 오래된 요청을 제거하고 새로운 요청을 수락합니다.