[Spring] 비동기 처리 @Async 적용 (2) - Spring 제공 비동기 처리?

spring 프레임워크에서 제공하는 비동기 처리 기능에는 여러 가지가 있다. 이번 글에서는 spirng에서 제공하는 비동기 처리에 대해 간단하게 정리해보려 한다.  

---

📚 @Async

• Spring의 전통적인 비동기 처리 방식 제공
• @Async를 메서드에 붙이면, 해당 메서드는 별도의 스레드에서 실행된다.
  • 이를 통해 메소드의 실행이 비동기적으로 이루어진다.
• @Async를 사용하면 메서드의 반환 값은 Future 또는 CompletableFuture 객체를 통해서 제공된다.
  • 해당 객체를 통해 비동기 작업의 결과를 얻거나 비동기 작업이 완료될 때까지 기다릴 수 있다.
• @EnableAsync를 설정 클래스에 추가하면 @Async가 붙은 메서드를 별도의 스레드에서 실행될 수 있도록 Spring이 자동으로 처리.

Future

- Java에서 제공하는 인터페이스로 비동기 계산의 결과를 나타낸다.

- Future를 사용하면 메인 스레드가 비동기 작업의 완료를 기다리지 않고 다음 작업을 수행할 수 있다.

- 이 후에 필요할 때 Future의 get() 메서드를 호출하여 비동기 작업의 결과를 가져올 수 있다. 

 

Future 제공 메서드

1. get()

   - 비동기 작업의 결과를 가져온다.

   - 만약 결과가 아직 준비되지 않았다면 결과가 준비될 때까지 블로킹(Blocking)된다.

            → 비동기 작업의 결과를 반환하는데, 만약 비동기 작업이 아직 완료 되지 않았다면, 해당 메서드를 호출한 스레드는 작업이 완료될 때까지 대기 상태가 되며 다른 작업을 진행하지 못하고 멈춰있다는 의미!

 

2. get(log timeout, TimeUnit unit)

   - 지정된 시간 동안 비동기 작업의 결과를 기다린다. 

   - 지정된 시간이 지나도 결과가 준비되지 않은 경우 TimeoutException을 던진다.

 

3. isDone()

   - 비동기 작업이 완료되었는지 여부를 반환한다.

 

4. cancel(boolean mayInterruptIfRunning)

   - 비동기 작업을 취소한다.

   - mayInterruptIfRunning 파라미터가 true인 경우, 작업을 수행 중인 스레드를 중단할 수 있다. 

 

📚 TaskExecutor 인터페이스

• Java의 Executor 인터페이스를 확장한 Spring의 인터페이스이다.
• execute(Runnable task) 메서드를 제공하여 주어진 작업을 별도의 스레드에서 실행하게끔 한다.
• 스레드 생성과 관리를 직접 처리하지 않고도 원하는 작업을 별도의 스레드에서 실행할 수 있다.
• 보통 @Async와 함께 사용되어 비동기 작업을 처리할 스레드 풀을 제공한다.  

 

📚CompletableFuture

• Java 8에서 도입된 클래스로, Future와 CompletionStage를 구현하고 있다. 
• 기존 Future에서는 비동기 작업 후에 수행해야 할 추가 작업을 지정하거나, 여러 비동기 작업을 조합해서 직접 구현하는 등의 문제점이 있었다.
  • Future의 get() 메서드는 결과가 준비될 때까지 블로킹되는 문제가 있었다. 
  • 만약 데이터베이스에서 큰 파일을 읽는 작업이 비동기적으로 수행되고 있고 이 작업의 결과를 Future 객체를 통해 받았다면?
    
    이때 Future.get() 메서드를 호출하면 위 파일이 완전히 읽힐 때까지 현재 스레드는 대기 상태!!
    
    즉, 파일 읽기 작업이 완료되지 않았다면 Future.get() 메서드는 반환하지 않고 대기 상태이기 때문에 현재 사용하는 스레드는 다른 어떤 작업도 진행하지 못하고 멈춰있게 된다. 
• Future의 대기 상태 문제를 해결하기 위해 비동기 작업이 완료될 때까지 블로킹되지 않는 방식으로 비동기 작업의 결과를 얻을 수 있게 하기 위해 나온 것.
• thenApply, thenAccept, thenCompose, thenCombine 등의 메서드를 제공하여 비동기 작업의 결과를 처리하거나 여러 비동기 작업을 연결하거나 조합하는 등의 작업 흐름을 조금 더 쉽게 구현할 수 있다. 

📚 WebFlux

• Spring 5에서 도입된 리액티브 웹 프레임워크.
• WebFlux는 비동기와 논블록킹 방식을 사용하여 요청을 처리한다.
  • 각 요청을 별도의 스레드에서 처리하는 전통적인 방식에 비해 스레드 사용을 최소화하고, 서버 리소스를 효율적으로 활용할 수 있게 해 준다. 
• Reactor의 객체인 Mono와 Flux를 이용하여 데이터 스트림을 표현한다. 
  • Reactive Streams 표준을 구현하며 데이터의 생성과 소비가 비동기적으로 이루어진다.
  • 높은 동시성을 필요로 하는 환경에서도 높은 성능과 효율을 달성할 수 있다.
• 즉, 적은 양의 스레드와 최소한의 하드웨어 자원으로 동시성을 핸들링하기 위해 만들어졌다. 
• WebFlux에서 사용하는 reactive library가 Reactor이고 Reactor가 Reactive Streams의 구현체이다.

🤔Mono, Flux란?

• Mono와 Flux의 차이점은 다루는 데이터의 개수이다. 

1. Mono

• 0 또는 1개의 결과를 반환하는 비동기 작업
• DB에서 단일 레코드를 조회하거나 HTTP 요청을 통해 단일 리소스를 가져오는 작업을 표현할 수 있다.
• 작업이 완료되면 결과를 '발행'하거나 정상적으로 완료되지 않은 경우 에러를 '발행'한다.

Mono.fromCallable(System::currentTimeMillis)
    .flatMap(time -> Mono.first(serviceA.findRecent(time), serviceB.findRecent(time)))
    .timeout(Duration.ofSeconds(3), errorHandler::fallback)
    .doOnSuccess(r -> serviceM.incrementSuccess())
    .subscribe(System.out::println);

 

2. Flux

• 0개 이상의 결과를 반환하는 비동기 작업
• DB에서 여러 레코드를 조회하거나 HTTP 요청을 통해 여러 리소스를 가져오는 작업을 표현할 수 있다. 
• Flux가 다루는 작업이 진행되는 동안 Flux는 결과를 계속해서 '발행' 하며 모든 결과가 '발행'되면 완료 신호를 '발행'하거나 정상적으로 완료되지 않은 경우 에러를 '발행'한다. 

Flux.fromIterable(getSomeLongList())
    .mergeWith(Flux.interval(100))
    .doOnNext(serviceA::someObserver)
    .map(d -> d * 2)
    .take(3)
    .onErrorResume(errorHandler::fallback)
    .doAfterTerminate(serviceM::incrementTerminate)
    .subscribe(System.out::println);

 

비동기 작업의 결과가 단일 값인 경우 : Mono

비동기 작업의 결과가 여러 값인 경우 : Flux

 

❓'발행'의 의미?

• 리액티브 프로그래밍에서 주로 사용되는 용어로 데이터를 전달하는 행위를 말한다. 
• 리액티브 스트림의 특성상 데이터가 연속적으로 생성되고 이를 구독자에게 순차적으로 전달하는 방식을 갖기 때문이다. 

 

참고

https://github.com/reactor/reactor-core