CompleteableFuture 를 이용한 비동기 처리

요즘 제작되는 어플리케이션은 대부분 네트워크 작업을 필요로 하는 경우가 늘어나고 있습니다.

단순 정보만 요청하고 받던 클라이언트 기반 프로그램부터 시작하여, 인터넷이 필요가 없을 것 같은 메모나 사진 촬영 앱도 공유 혹은 클라우드 처리를 지원하기 때문에 네트워크 작업이 없는 경우는 거의 없다고 생각할 수 있을 것 같습니다.

이러한 네트워크와 연결을 시도하는 프로그램을 제작해본 사람들이라면 알고 있겠지만, 네트워크 처리나 File IO 등 오래걸리는 일은 비동기 처리를 해야 합니다. 

비동기 처리는 하는 이유는 IO 작업이 일어나는 동안 메인스레드(아마도 UI 스레드)가 그동안 놀고 있는 상태(블럭 상태 - CPU 사이클이 낭비됨.)를 피하기 위해서입니다. 사용자 측면에서 어떤 정보를 읽어오는 동안 UI 가 멈춘다면, 불편하다고 충분히 느낄 수 있을 것입니다. ㅡㅡ^

아래는 비동기 처리 작업에 대한 그림입니다. 

비동기 프로세스(Async) 의 경우 Process B 가 실행되는 동안 Process A 는 계속해서 작업을 할 수 있습니다.

Java7 부터는 Future 인터페이스를 통해서, 이러한 비동기 프로세스를 수행할 수 있었습니다.

ExecutorService executorService = Executors.newCachedThreadPool();
 
Future<Double> future = executorService.submit(()-> {
    // 편의 상 람다 사용. 자바7 에서는 사용 불가.
    Thread.sleep(2000);
 
    return 1000.0;
});
 
System.out.println("비동기 처리를 하는 동안 다른 일처리.");
 
try {
    // 타임아웃 3초로 지정.
    System.out.println("결과 : " + future.get(3000, TimeUnit.MILLISECONDS));
} catch (Exception e) {
    e.printStackTrace();
}
 
// 출력 결과
// 비동기 처리를 하는 동안 다른 일처리.
// 결과 : 1000.0

Future 클래스는 비동기 계산이 끝났는지 확인할 수 있는 isDone, 타임아웃 기간을 결정하고 결과를 출력하는 get 메소드 등이 있습니다.

간단히 비동기 처리는 되지만 조금 아쉽습니다. 

실무에서는 비동기 처리가 꼭 하나씩 생긴다고 볼 수 없으며, 각 비동기 처리에 대한 결과를 동기를 맞춰 또 다른 결과를 내야할 수 도 있습니다. 즉 각 Future 클래스 간 여러 의존성에 대한 관리가 힘들 수 있습니다.

JAVA8 에서는 복잡한 비동기처리를 선언형으로 이용할 수 있는 CompleteableFuture 를 제공하며, Stream API 나 Optional 같이 람다표현식과 파이프라인을 사용하여 비동기 작업을 조합할 수 있습니다.

일단 CompleteableFuture 의 간단한 예제는 아래와 같습니다. CompleteableFuture 는 기본적으로 supplyAsync, runAsync 등 팩토리 메소드를 제공하며, 쉽게 비동기 작업을 수행할 수 있습니다.

Future<Double> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (Exception e){
                e.printStackTrace();
            }
 
            return 1000.0;
        });
 
System.out.println("비동기 처리를 하는 동안 다른 일처리.");
 
try {
    // 타임아웃 3초로 지정.
    System.out.println("결과 : " + future.get(3000, TimeUnit.MILLISECONDS));
} catch (Exception e) {
    e.printStackTrace();
}

비동기로 처리되어야 할 일이 많아지면 어떨까요? 동시에 앞의 예제처럼 Sleep 을 해야하는 task 가 다수일 때는 간단하게 두가지의 선택 경로를 생각해볼 수 있습니다.

Supplier<Double> supplier = () -> {
    try {
        Thread.sleep(2000);
    } catch (Exception e){
        e.printStackTrace();
    }
 
    return 1000.0;
};
 
List<Supplier<Double>> supplierList = Arrays.asList(supplier, supplier, supplier, supplier);
 
// 병렬 스트림을 이용. 각 태스크를 병렬로 하여 성능을 높이자.
supplierList.parallelStream().
    map(Supplier::get).
    reduce(Double::sum).
    ifPresent(System.out::println);
 
// CompletableFuture 를 이용한 비동기적으로 처리
{
    List<CompletableFuture<Double>> completableFutures = supplierList.stream().
            map(CompletableFuture::supplyAsync).
            collect(Collectors.toList());
 
    // join 메소드는 모든 비동기 동작이 끝나길 기다립니다.
    completableFutures.stream().
            map(CompletableFuture::join).
            reduce(Double::sum).
            ifPresent(System.out::println);
}

두 구현 방식에 따라 결과는 큰 차이가 나지 않을 수 있습니다. 

그러나 일반 순차 Stream 을 병렬 Stream 으로 변경한 첫 번째 방법이 간단해 보입니다.

굳이 CompletableFuture 를 쓸 필요가 없어보이지만, 병렬스트림과 달리 이를 이용한 방법은 executor 를 커스터마이징 할 수 있습니다.

List<Supplier<Double>> supplierList = IntStream.range(0, 100).mapToObj(n -> supplier).collect(Collectors.toList());
 
supplierList.parallelStream().
        map(Supplier::get).
        reduce(Double::sum).
        ifPresent(System.out::println);
 
// CompletableFuture 를 이용한 비동기적으로 처리
{
 
    final Executor executor = Executors.newFixedThreadPool(Math.min(supplierList.size(), 100), r -> {
                Thread t = new Thread(r);
                // 데몬 스레드 정의
                // 일반 스레드가 실행 중일 때 자바 프로그램은 종료되지 않음 -> 어떤 이벤트를 한없이 기다리면서 종료되지 않은 일반 자바 스레드가 있으면 문제
                // 데몬 스레드는 자바 프로그램이 종료될 때 종료
                t.setDaemon(true);
                return t;
            });
 
    List<CompletableFuture<Double>> completableFutures = supplierList.stream().
            map(CompletableFuture::supplyAsync).
            collect(Collectors.toList());
 
    // join 메소드는 모든 비동기 동작이 끝나길 기다립니다.
    completableFutures.stream().
            map(CompletableFuture::join).
            reduce(Double::sum).
            ifPresent(System.out::println);
}
 
// 병렬스트림 걸린 시간 : 26066초
// CompletableFuture 사용 걸린 시간 : 2015초

놀라운 결과입니다. 걸린 시간 자체가 무려 10배가 넘게 차이가 남을 알 수 있습니다.

즉 로직에 따라 Executor 를 다르게 하여, 최적화 시키는 것이 효과적일 수 있음을 알 수 있습니다.

비동기 처리의 최적화와 더불어, CompleteableFuture 는 람다표현식이나 파이프라인 메소드를 이용하여, 비동기 연산을 조합할 수 있습니다. 다음 포스팅에서는 이에 대해 다루어 보도록 하겠습니다. 

2017/03/24 - [개발이야기/함수형 방법론] - CompleteableFuture 를 이용한 비동기 처리 조합

자바 8 인 액션 국내도서 저자 : 라울-게이브리얼 우르마(RAOUL-GABRIEL URMA),마리오 푸스코(MARIO FUSCO),앨런 마이크로프트(ALAN MYCROFT) / 우정은역 출판 : 한빛미디어 2015.04.01 상세보기