[Java] 비동기 처리를 위한 CompletableFuture 완벽 가이드

목차

1. 비동기 프로그래밍의 개념
2. 왜 CompletableFuture를 사용할까?
3. CompletableFuture 사용법

 

비동기 작업을 위해 Java8부터 도입된 CompletableFuture에 대해 살펴보자.

 

1. 비동기 프로그래밍의 개념

CompletableFuture에 대해 설명하기 앞서, 동기와 비동기 그리고 블로킹과 논블로킹에 대한 개념을 짚고 넘어가자. 왜 이 개념에 대해 짚고 넘어가야할까? 

동기와 비동기는 실행 시점에 대한 개념으로 어떻게 시작하고 진행할 지에 대한 것이며, 블로킹 논블로킹은 결과를 받는 방식과 관련된 개념으로 어떻게 기다리고 처리하는지에 대한 것이다. 이 개념을 시작으로 자바의 비동기 프로그래밍에 다가갈 수 있다.

 

1.1 동기 vs 비동기

• 동기: 한 작업이 완료된 후에 다음 작업을 시작하는 것이다.
• 비동기: 작업의 완료를 기다리지 않고 다음 작업을 실행하는 것이다.

 

1.2 블로킹 vs 논블로킹

• 블로킹: 작업이 완료될 때까지 스레드가 대기하는 것이다.
• 논블로킹: 작업의 완료와 관계없이 스레드가 다른 작업을 수행할 수 있는 것이다.

✅CompletableFuture는 thenApply, thenAccept 등을 제공하여 비동기-논블로킹 방식을 효율적으로 구현할 수 있도록 해준다.

 

2. 왜 CompletableFuture를 사용할까?

2.1 자바의 비동기 처리 변화

자바에서 비동기 프로그램은 다음과 같은 발전을 거쳐왔다.

(1) Thread: 초기 비동기 처리 방식이다.

• 예시
  

  new Thread(() -> {
      // 비동기 작업 수행
      System.out.println("별도 스레드에서 작업 수행");
  }).start();

(2) ExcecutorService: Java 5부터 ExecutorService와 Future도입 되었다.

• 예시
  

  ExecutorService executor = Executors.newFixedThreadPool(10);
  Future<String> future = executor.submit(() -> {
      // 비동기 작업 수행
      return "작업 결과";
  });
  String result = future.get(); // 블로킹 호출

(3) CompletableFuture: Java 8부터 함수형 스타일의 비동기 처리 도입 되었다.

• 예시
  

  CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
      // 비동기 작업 수행
      return "작업 결과";
  });
  future.thenAccept(result -> System.out.println("결과: " + result));

 

2.2 CompletableFuture가 해결하는 것

(1) Future의 한계

Java 5의 Future 인터페이스는 다음과 같은 한계가 있었다.

• 블로킹 방식의 get() 메서드만 제공
• 여러 Future를 조합할 수 없음
• 제한적인 예외 처리
• 콜백 등록 불가능

(2) CompletableFuture가 제공하는 기능

CompletableFuture는 Future의 한계를 보완하고 다음과 같이 다양한 기능을 제공한다.

• 논블로킹 방식 제공
• 함수형 프로그래밍 스타일 지원
• 체이닝 가능 (thenApply, thenAccept, thenRun 등)
• 여러 작업들을 조합 가능 (thenCombile, thenCompose, allOf, anyOf 등)
• 다양한 예외처리 제공 (exceptionally, handle 등)
• complete()를 통한 명시적 완료
• 타임아웃 기능 내장

 

3. CompletableFuture 사용법

3.1 비동기 처리

• runAsync: 반환값이 없는 비동기 작업 실행에 사용할 수 있다.
  • 예시
    

    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
        try {
            Thread.sleep(1000);
            System.out.println("비동기 작업 실행 중... " + Thread.currentThread().getName());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    });
    
    // 작업 완료 대기
    future.join();
    System.out.println("runAsync 작업 완료");

• supplyAsync: 반환값이 있는 비동기 작업 실행에 사용할 수 있다.
  • 예시
    

    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(1000);
            System.out.println("비동기 작업 실행 중... " + Thread.currentThread().getName());
            return "작업 결과";
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    });
    
    // 결과 가져오기 (블로킹)
    String result = future.join();
    System.out.println("supplyAsync 작업 결과: " + result);

• Executor와 같이 사용할 경우
  • 기본적으로 CompletableFuture는 ForkJoinPool.commonPool()을 사용하지만, 원하는 경우 커스텀 Executor를 지정할 수 있다. ForkJoinPool의 공용 스레드풀은 시스템의 가용 프로세서 수에 따라 자동으로 결정되는데, 이는 CPU 집약적인 작업에 최적화되어 있다. 반면에, 많은 양의 I/O 작업이나 blocking 작업을 처리할 때는 커스텀 Executor를 사용하는 것이 더 효율적이다.
  • 예시
    

    ExecutorService executor = Executors.newFixedThreadPool(10);
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("현재 스레드: " + Thread.currentThread().getName());
        return "커스텀 스레드풀에서 실행";
    }, executor);
    
    String result = future.join();
    System.out.println(result);

 

3.2 콜백

• thenApply: 이전 작업의 결과를 받아 변환하여 새로운 값을 반환한다.
  • 예시
    

    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "hello");
    CompletableFuture<Integer> lengthFuture = future.thenApply(s -> s.length());
    System.out.println("문자열 길이: " + lengthFuture.join()); // 출력: 문자열 길이: 5

• thenAccept: 이전 작업의 결과를 받아 작업을 수행하고 결과를 반환하지 않는다.
  • 예시

    CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> "hello")
        .thenAccept(result -> System.out.println("결과: " + result));
    future.join(); // 출력: 결과: hello

• thenRun: 이전 작업 완료 후 실행할 작업을 지정하는 것으로 이전 작업의 결과를 사용하지 않는다.
  • 예시
    

    CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> "hello")
        .thenRun(() -> System.out.println("작업 완료됨"));
    future.join(); // 출력: 작업 완료됨

 

3.3 조합

• allOf: 모든 작업이 완료될 때까지 기다리고, 모든 작업이 완료되면 CompletableFuture<Void>를 반환한다.
  • 예시
    

    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "결과 1");
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "결과 2");
    CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "결과 3");
    
    CompletableFuture<Void> allFutures = CompletableFuture.allOf(future1, future2, future3);
    
    allFutures.thenApply(v -> {
        List<String> results = Stream.of(future1, future2, future3)
            .map(CompletableFuture::join)
            .collect(Collectors.toList());
        return results;
    });

• anyOf: 여러 CompletableFuture 중 어느 하나라도 완료되면 결과를 반환한다.
  • 예시
    

    CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(future1, future2, future3);
    anyFuture.thenAccept(result -> System.out.println("첫 번째 완료된 결과: " + result));

• thenCompose: 이전 작업의 결과를 받아 새로운 작업을 수행하는 것이다.
  • 예시
    

    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "user_id")
        .thenCompose(userId -> fetchUserDetails(userId));

• thenCombine: 두 작업의 결과를 받아 조합하는 것이다.
  • 예시
    

    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World");
    
    CompletableFuture<String> combinedFuture = future1.thenCombine(future2, (s1, s2) -> s1 + " " + s2);

 

3.4 예외처리

• exceptionally: 예외가 발생할 경우에 대체 값을 제공할 수 있다.
  • 예시
    

    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        if (Math.random() > 0.5) {
            throw new RuntimeException("에러 발생!");
        }
        return "성공";
    })
    .exceptionally(ex -> {
        System.out.println("예외 발생: " + ex.getMessage());
        return "기본값";
    });

• handle: 정상적인 결과와 예외를 모두 매개변수로 받아 처리할 수 있다.
  • 예시
    

    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        if (Math.random() > 0.5) {
            throw new RuntimeException("에러 발생!");
        }
        return "성공";
    })
    .handle((result, ex) -> {
        if (ex != null) {
            System.out.println("예외 발생: " + ex.getMessage());
            return "에러 시 기본값";
        }
        return result + " 처리 완료";
    });

• whenComplete: 작업 완료 시 부가 작업을 처리할 수 있다.
  • 예시
    

    future.whenComplete((result, ex) -> {
        if (ex != null) {
            System.out.println("작업 실패: " + ex.getMessage());
        } else {
            System.out.println("작업 성공 결과: " + result);
        }
    });

 

 

정리하자면 CompletableFuture는 자바에서 비동기 프로그래밍을 간결하고 함수형으로 작성할 수 있도록 다양한 기능을 제공한다. 비동기 작업의 흐름을 구성하고, 다양한 비동기 작업을 조합하거나 예외 처리 등을 다양하게 구현할 수 있다.

 

📒참고자료
https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/CompletableFuture.html
https://11st-tech.github.io/2024/01/04/completablefuture/
https://www.baeldung.com/java-executorservice-vs-completablefuture