Future는 비동기 처리 결과를 얻기 위해 주기적으로 상태를 확인해야 하며, 이로 인해 CPU 사용률이 증가할 수 있습니다.

ExecutorService executorService = Executors.newFixedThreadPool(10);
Future<String> asyncResult = executorService.submit(() -> {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "asyncResult";
});
while (true) {
    if (asyncResult.isDone()) {
        System.out.println(asyncResult.get());
        break;
    }
}

CompletableFuture는 관찰자 패턴을 기반으로 한 비동기 처리 방식으로, 결과를 블록킹 방식으로 가져올 수 있으며 성능 최적화가 가능합니다.

System.out.println("=============CompletableFuture===================");
CompletableFuture<String> asyncTask1 = CompletableFuture.supplyAsync(() -> "데이터1")
    .thenApply(result -> {
        System.out.println("asyncTask1 후처리: " + result);
        return "데이터2";
    });
System.out.println(asyncTask1.get());

System.out.println("=============CompletableFuture===================");
CompletableFuture<String> asyncTask2 = CompletableFuture.supplyAsync(() -> "데이터1")
    .thenAccept(result -> {
        System.out.println("asyncTask2 후처리: " + result);
    });
System.out.println(asyncTask2.get());

CompletableFuture 활용 사항

• API 문서: Flink Async I/O 가이드
• 메서드 분류
  • runAsync: 반환값 없음
  • supplyAsync: 반환값 있음
  • thenAccept: 반환값 없음
  • thenApply: 반환값 있음
  • thenRun: 이전 결과 무관한 다음 작업 실행
  • get(): 타임아웃 설정 가능하여 장시간 블록킹 방지
• 구현 예시

  
  class AsyncDBHandler extends RichAsyncFunction<String, Tuple2<String, String>> {
      private transient DBClient dbClient;
  
      @Override
      public void open(Configuration parameters) throws Exception {
          dbClient = new DBClient(host, port, credentials);
      }
  
      @Override
      public void close() throws Exception {
          dbClient.close();
      }
  
      @Override
      public void asyncInvoke(String key, ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {
          Future<String> queryResult = dbClient.query(key);
          CompletableFuture.supplyAsync(() -> {
              try {
                  return queryResult.get();
              } catch (Exception e) {
                  return null;
              }
          }).thenAccept(dbValue -> {
              resultFuture.complete(Collections.singletonList(new Tuple2<>(key, dbValue)));
          });
      }
  }
  
  DataStream<String> inputStream = ...;
  DataStream<Tuple2<String, String>> output = AsyncDataStream.unorderedWait(
      inputStream, new AsyncDBHandler(), 500, TimeUnit.MILLISECONDS, 50);
  

참고 사항

• 타임아웃 설정: 요청이 지정된 시간 내에 완료되지 않으면 자동 취소됨
• 동시 처리 수 제한: 병렬 요청 수를 제어하여 시스템 부하 최소화
• 타임아웃 처리: 기본적으로 예외 발생 시 재시도되나, timeout() 메서드 오버라이드로 커스터마이징 가능
• 결과 순서
  • unorderedWait: 비순차적 결과 전달
  • orderedWait: 요청 순서와 동일한 결과 순서 유지