[WebFlux] 비동기/동기, non-blocking/blocking, CompletableFuture

팀에서 IoT를 비동기적으로 처리하기 위해 WebFlux를 사용하고 있다.

WebFlux에 대해 얕게만 알고 사용했기 때문에 이번 기회에 공부해보기로 한다. 

 

✴️ Sync(동기)와 Async(비동기)

🔶 Sync : 클라이언트에서 메서드 호출 → 메서드 제공하는 곳에서 기능을 모두 수행해 결과값이 결정되면 그때 반환
👉 요청한 작업에 대해 완료 여부를 신경 써서 작업을 순차적으로 수행할지 아닌지 결정

🔶 Async : 결과값이 결정되기 전에 일단 반환 (순서 보장 X)
👉 요청한 작업에 대해 완료 여부를 따지지 않고 자신의 다음 작업을 그대로 수행

 

✴️ Blocking과 Non-Blocking

🔷 Blocking : 요청한 작업이 끝날 때까지 다른 작업하지 않고 기다림

🔷 Non-Blocking : 요청한 작업이 수행되는 동안 다른 작업 가능. 

 

✴️ 콜백함수 

👉 다른 작업의 완료 여부나 결과에 대한 후처리를 위해 이용

 

 

CompletableFuture 란? 

java.util.concurrent 패키지에 속하는 클래스 중 하나로, 비동기 프로그래밍을 보다 쉽게 구현할 수 있도록 도와준다. 

Java 8에 처음 도입되었고, Future와 CompletionStage을 구현하고 있다. 

public class CompletableFuture<T> implements Future<T>, CompletionStage<T>

 

Future와 CompletionStage는 뭘까.

 

 

Future

• Java 5에 도입
• 비동기적인 작업 수행
• 작업 상태 확인 - 완료, 취소, 진행 중 여부
• 해당 작업이 완료되면 결과를 반환하는 인터페이스

🔶 주요 메서드 

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    
    boolean isCancelled();
    
    boolean isDone();
    
    V get() throws InterruptedException, ExecutionException;
    
    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

 

• cancel : 작업 취소 👉 mayInterruptIfRunning이 true이면 실행 중인 작업 중단 
• get() : 작업이 완료될 때까지 기다린 후 결과 반환 👉 작업이 완료될 때까지 호출 스레드는 블로킹됨

---

🔶 한계

• cancel을 제외하고 외부에서 future 컨트롤 X 👉 콜백 미지원
• 블로킹 : Future.get() 메서드는 결과가 준비될 때까지 블로킹되므로 비동기 작업의 이점 충분히 활용 X
• 반환된 결과를 get()해서 접근하기 때문에 비동기 처리 어려움
• 완료되거나 에러 발생했는지 구분 어려움 
• 조합 어려움 : 여러 비동기 작업을 조합, 순차적으로 실행, 결과 조합 어려움 

 

 

CompletionStage

• Java 8에 도입 
• 비동기적인 작업 수행
• 해당 작업이 완료되면 결과를 처리하거나 다른 CompletionStage를 연결하는 인터페이스
• 비동기 task들을 실행하고 값을 변형하는 등 chaining을 이용한 조합 가능
• 에러를 처리하기 위한 콜백 제공

 

🔶 주요 메서드 

public interface CompletionStage<T> {
    public <U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn);
    public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U> fn);
    
    public CompletionStage<Void> thenAccept(Consumer<? super T> action);
    public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
    
    public CompletionStage<Void> thenRun(Runnable action);
    public CompletionStage<Void> thenRunAsync(Runnable action);
    
    public <U> CompletionStage<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn);
    public <U> CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn);
    
    public CompletionStage<T> exceptionally(Function<Throwable, ? extends T> fn);
}

• thenAccept[Async] 
  • Consumer를 파라미터로 받음 
  • 현재 단계의 결과를 사용하여 동기적으로 작업 수행하고 반환 X
  • 값을 받아서 action만 수행하는 경우에 유용

 

• thenApply[Async]
  • Function을 파라미터로 받음
  • 이전 task로부터 T 타입의 값을 받아서 가공하고 U 타입의 값 반환
  • 다음 task에게 반환했던 값이 전달됨
  • thenApply → 현재 CompletableFuture를 완료한 동일한 스레드를 사용하여 변환 함수 실행
    👉 변환 함수의 실행은 결과가 사용 가능해진 직후에 발생 
  • CompletableFuture의 결과에 대해 체이닝 작업 수행 
  • 값을 변형해서 전달해야 하는 경우 유용

 

• thenCompose[Async]
  • Fuction을 파라미터로 받음
  • 이전 task로부터 T 타입의 값을 받아서 가공하고 U 타입의 CompletionStage 반환
  • 반환한 CompletionStage가 done 상태가 되면 값을 다음 task에 전달
  • 다른 future를 반환해야하는 경우 유용 

 

• thenRun[Async]
  • Runnable을 파라미터로 받음
  • 이전 task로부터 값을 받지 않고 값을 반환 x
  • 다음 task에게 null이 전달됨
  • future가 완료되었다는 이벤트를 기록할 때 유용 

 

• then*[Async]
  • 비동기적으로 실행
  • Async 👉 별도의 스레드에서 실행 
  • 호출된 스레드는 즉시 반환되고 결과는 나중에 사용 가능
  • Executor 사용해서 스레드 풀을 통해 스레드 관리 가능

 

• exceptionally
  • Function을 파라미터로 받음
  • 이전 task에서 발생한 exception을 받아서 처리하고 값 반환
  • 다음 task에게 반환된 값을 전달
  • future 파이프에서 발생한 에러를 처리할 때 유용 

 

 

 

CompletableFuture

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) { … }
    public static CompletableFuture<Void> runAsync(Runnable runnable) { … }
    
    public boolean complete(T value) { … }
    public boolean isCompletedExceptionally() { … }
    
    public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) { … }
    public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) { … }
}

 

• supplyAsync
  • 비동기적으로 값을 계산하고 그 값을 반환 
  • Supplier를 제공하여 CompletableFuture 생성 가능
  • Supplier의 반환값이 CompletableFuture의 결과로

 

• runAsync
  • 비동기적으로 실행되지만 반환값이 없는 작업 실행 
  • Runnable 제공하여 CompletableFuture 생성 가능
  • 다음 task에 null 전달됨 

🔷 Supplier 인터페이스 

@FunctionalInterface
public interface Supplier<T> {
    T get();
}​

 

매개변수 없이 어떤 값을 제공하는 메서드를 정의하는 함수형 인터페이스
👉 값을 생성하거나 계산하는 데 사용됨

 

🔷 Runnable 인터페이스

@FunctionalInterface
public interface Runnable {
    void run();
}​

 

매개변수가 없고 반환값이 없는 작업을 정의하는 함수형 인터페이스
👉 작업을 실행하는 데 사용됨 

 

• complete
  • complete에 의해서 상태가 바뀌었다면 true, 아니라면 false 반환

 

• isCompletedExceptionally
  • CompletableFuture 객체가 완료된 상태
  • exception에 의해서 complete 되었는지 확인 
  • 완료된 결과가 예외를 던지면 true 반환, 아니면 false 반환
  • 예외를 던지지 않고 정상적으로 결과를 반환했을 경우 false 반환
  • 비동기 작업이 예외를 던지는지 여부 확인할 때 유용하게 사용됨 

 

• allOf
  • 여러 completableFuture를 모아서 하나로 변환
  • 모든 completableFuture가 완료되면 상태가 done으로 변경
  • void 반환하므로 각각의 값에 get으로 접근

 

• anyOf
  • 여러 completableFuture를 모아서 하나로 변환
  • 주어진 future 중 하나라도 완료되면 상태가 done으로 변경
  • 제일 먼저 done 상태가 되는 future의 값 반환

---

🔶 한계

• 지연 로딩 기능 제공 X
• 지속적으로 생성되는 데이터 처리하기 어려움

 

 

 

 

 

참고자료 

https://www.baeldung.com/java-completablefuture

https://11st-tech.github.io/2024/01/04/completablefuture/

https://www.youtube.com/watch?v=oEIoqGd-Sns

https://dkswnkk.tistory.com/733