Java Async - CompletableFuture

 

이번에는 자바 8부터 비동기를 기능을 지원해주는 CompletableFuture에 대해 기록하려합니다. 

이 글은 https://www.youtube.com/watch?v=PzxV-bmLSFY&t=8s토비좌 의 강의를 듣고 공부를 위해 작성하고 있습니다.

 

기존에 자바에서 비동기 작업을 진행한다 하면 별도의 쓰레드풀 생성, Future, FutureTask 을 통한 비동기처리 또는 ThreadPoolTaskExecutor등 다양한 방법이 있긴했습니다.

 

하지만 8에서 보다 간단하면서  비동기 작업의 결과를 만들어 낼 수 있는 CompletableFuture가 등장했습니다.

 

특징으로는

1. CompletionStage 을 implement하고있어 비동기 작업을 의존적으로 또다른 기능을 수행할 수 있게 해줍니다. 그리고 2. 별도 thread pool 생성 없이도 8부터는 ForkjoinPool의 CommonPool을 사용하게 되어 추가로 별도의 쓰레드 생성하는 코드를 사용하지 않아도 됩니다.

3. 파이트라인식으로 코드를 사용할 수 있어 직관적으로 코드를 볼 수 있습니다.

4. async 의 작업이 여러개가 필요할 떄, 모든 값이 완료될때까지 기다릴지 통과할지 선택할 수 있는 기능을 지원합니다.

 

이외에도 제가 모르는 장점이 많이 있는것으로 알고있습니다.(해당부분은 알게되면 계속 업데이트 하려합니다.)

 

 

1. 리턴이 없는 사용 예시

@Slf4j
public class DemoApplication {
    private static final Logger log = LoggerFactory.getLogger(DemoApplication.class);
    public static void main(String[] args) throws IOException, InterruptedException {
        CompletableFuture
                .runAsync(() -> log.info("runAsync"))
                .thenRun(() -> log.info("thenRun"))
                ;
        log.info("exit");
    }
}    
    

다음 코드와 결과를 보면 exit 자체는 mainThread 를 사용하였으나, CompletableFuture 에서 사용된 RunAsync, thenRun()의 경우 별도의 쓰레드(ForkJoinPool 에서 생성된 같은 쓰레드 사용)를 사용하는 것을 볼 수 있습니다.

runAsync

별도의 리턴이 없는 비동기 실행 방법, 간단하게 람다식으로 사용가능.

thenRun

최초 시작된 Async작업 이후, 이어서 시작시킬 작업 입력을 지원 ( 기본적으로는 runAsync에서 사용했던 스레드와 같은 스레드에서 작업 진행)

 

2. 리턴을 하는 경우에 대한 사용 예시

@Slf4j
public class DemoApplication {
    private static final Logger log = LoggerFactory.getLogger(DemoApplication.class);
    public static void main(String[] args) throws IOException, InterruptedException {
        CompletableFuture
                .supplyAsync(() -> {
                    log.info("runAsync");
                    //if (1 == 1) throw new RuntimeException();
                    return 1;
                })
                .thenCompose(s -> {
                    log.info("thenCompose1 {}", s);
                    return CompletableFuture.completedFuture(s + 1);
                })
                .thenCompose(s -> {
                    log.info("thenCompose2 {}", s);
                    return CompletableFuture.completedFuture(s + 1);
                })
                .thenApply(s -> {
                    log.info("thenApply1 {}", s);
                    return s * 2;
                })
                .thenApply(s -> {
                    log.info("thenApply2 {}", s);
                    return s * 2;
                })
                //.exceptionally(e -> -10)
                .thenAccept(
                        s -> log.info("thenAccept {}", s)
                )
                .exceptionally(e -> {
                    log.info("exceptionally {}", e.getMessage());
                    return null;
                });
        log.info("exit");
    }
}
                

Async가 체인이 된다는것을 보이기 위하여 조금 길게 사용하였습니다.

 

 

리턴값이 필요한경우, supplyAsync라는 로직을 통해서 실행하게 됩니다. 이전 코드에서 봤던 thenAsync 외에도  thenCompose가 등장하는데, 해당 메서드는 일반 value가 아닌, CompletableFuture를 리턴해야하는 경우, 해당 값을 사용합니다. ( 사용예로는 별도의 비동기 작업을 진행 완료 후, 완료된 값을 CompletableFuture 에 담아 리턴하는 경우)

thenAccept 에서는 리턴값이 없으며, 최종적으로 넘겨받은 값에 대한 처리를 지원합니다. 

exceptionally 에서는 파이프라인으로 이어진 비동기 작업중, 에러를 잡아서 해당 값을 적용해줍니다.

 

 

만약 if (1 == 1) throw new RuntimeException(); 주석과 .exceptionally(e -> -10) 주석이 풀려있다면 어떻게 될까요?

강제로 에러를 던지게 되면 다음과 같이 .exceptionally(e -> -10) 에서 에러를 캐치하고 아래에 있는 에러는 잡히지 않게 됩니다.

그 이유로 먼저 exceptionally 에서 잡히고, thenAccept를 진행하기에, 맨 아래에 있는 exceptionally 는 에러가 아니게됩니다.

 

만약 thenAccept 에서 다시 에러를 던지게 되면 다음 사진과 같이 둘다 캡쳐되는것을 볼 수 있습니다.

이외에도 비동기의 작업을 합쳐서 다른 무언가의 기능을 지원해주는 thenCombine, Async pipeline에서 별도의 thread를 사용할 수 있게 해주는 기능 ( 별도의 thread pool 적용 필요) 등 여러 기능을 지원하고 있어, 기존과는 편한 비동기 작업을 지원해주고 있습니다.

 

 

얼른 운영관리 기능 코드에 테스트코드로 변경해봐야지