[Java] CompletableFuture

CompletableFuture란?

• Future와 CompletionStage를 구현한 클래스
• 비동기 처리를 쉽게 처리할 수 있는 방법을 제공한다.
• 여러 CompletableFuture를 병렬로 처리하거나 병합 처리, 취소, 에러 처리 등을 제공한다.

기본 사용법

CompletableFuture<String> future = new CompletableFuture<>();

new Thread(() -> {
    log("run");
    future.complete("Hello World");
}).start();

log("result : " + future.get());

[11:39:41.021406][Thread-0] run
[11:39:41.037362900][main] result : Hello World

completedFuture()

• 이미 활용할 데이터가 있을 경우 사용.

CompletableFuture<String> future = CompletableFuture.completedFuture("Hello World");

log("result : " + future.get());

[11:40:45.714630200][main] result : Hello World

cancel()

• cancel() 호출시 get()에서 CancellationException이 발생할 수 있어 예외처리를 해주어야 한다.

CompletableFuture<String> future = new CompletableFuture<>();

new Thread(() -> {
    log("run");
    future.cancel(false);
}).start();

try {
    log("result : " + future.get());
} catch (CancellationException e) {
    log("error : " + e);
}

[11:42:56.248613200][Thread-0] run
[11:42:56.263574100][main] error : java.util.concurrent.CancellationException

handle()

• 예외 처리를 위한 기능을 제공한다.

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> {
        log("future");
        throw new IllegalStateException("ERROR");
    })
    .handle((text, throwable) -> {
        log("error : " + throwable);
        return "Error";
    });

[11:48:37.049525100][ForkJoinPool.commonPool-worker-19] future
[11:48:37.066478500][ForkJoinPool.commonPool-worker-19] error : java.util.concurrent.CompletionException: java.lang.IllegalStateException: ERROR
[11:48:37.067475300][main] result : Error

supplyAsync()

• 직접 스레드 생성하지 않고 비동기 처리가 가능하게 해준다.
• 데이터를 리턴해주어야 할 때 사용한다.

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> {
        log("future");
        return "Hello World";
    });

log("result : " + future.get());

[11:43:45.796556500][ForkJoinPool.commonPool-worker-19] future
[11:43:45.811515900][main] result : Hello World

runAsync()

• supplyAynsc()와 동일
• 데이터를 리턴해줄 필요 없을 때 사용한다.

CompletableFuture<Void> future = CompletableFuture
    .runAsync(() -> {
        log("future");
    });

log("result : " + future.get());

[11:45:25.319738500][ForkJoinPool.commonPool-worker-19] future
[11:45:25.334797400][main] result : null

thenApply()

• supplyAsync()의 처리 결과를 활용해 후처리 작업을 해야할 때 사용한다.
• 결과값을 리턴해야할 경우 사용.

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> {
        sleep(1000);
        log("supplyAsync");
        return "Hello World";
    })
    .thenApply((text) -> {
        sleep(1000);
        log("thenApply");
        return text + "!";
    });

log("result : " + future.get());

[11:53:47.213973300][ForkJoinPool.commonPool-worker-19] supplyAsync
[11:53:48.233912400][ForkJoinPool.commonPool-worker-19] thenApply
[11:53:48.240899900][main] result : Hello World!

thenAccept()

• thenApply()와 동일.
• 결과값 리턴이 필요 없을 경우 사용.

CompletableFuture<Void> future = CompletableFuture
    .supplyAsync(() -> {
        sleep(1000);
        log("supplyAsync");
        return "Hello World";
    })
    .thenAccept((text) -> {
        sleep(1000);
        log("thenAccept");
    });

log("result : " + future.get());

[11:55:05.767468500][ForkJoinPool.commonPool-worker-19] supplyAsync
[11:55:06.789277100][ForkJoinPool.commonPool-worker-19] thenAccept
[11:55:06.791285600][main] result : null

thenCommpose()

• 여러 작업을 순차적으로 수행할 때 사용한다.

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> {
        sleep(1000);
        log("supplyAsync");
        return "Hello";
    })
    .thenCompose((text) -> CompletableFuture.supplyAsync(() -> {
        sleep(1000);
        log("thenCompose1");
        return text + " World";
    }))
    .thenCompose((text) -> CompletableFuture.supplyAsync(() -> {
        sleep(1000);
        log("thenCompose2");
        return text + " Java";
    }));

log("result : " + future.get());

[11:57:24.933529100][ForkJoinPool.commonPool-worker-19] supplyAsync
[11:57:25.959989][ForkJoinPool.commonPool-worker-5] thenCompose1
[11:57:26.972361900][ForkJoinPool.commonPool-worker-5] thenCompose2
[11:57:26.978944700][main] result : Hello World Java

thenCombine()

• 여러 작업을 동시에 수행할 때 사용한다.

CompletableFuture<String> future1 = CompletableFuture
    .supplyAsync(() -> {
        sleep(1000);
        log("future1");
        return "Hello";
    });

CompletableFuture<String> future2 = CompletableFuture
    .supplyAsync(() -> {
        sleep(1000);
        log("future2");
        return "World";
    });

CompletableFuture<String> future = future1.thenCombine(future2, (text1, text2) -> text1 + " " + text2);

log("result : " + future.get());

[12:01:06.897832500][ForkJoinPool.commonPool-worker-19] future1
[12:01:06.897832500][ForkJoinPool.commonPool-worker-5] future2
[12:01:06.919774300][main] result : Hello World

anyOf()

• 빨리 처리되는 1개의 결과만 가져오는 메소드

CompletableFuture<String> future1 = CompletableFuture
    .supplyAsync(() -> {
        sleep(3000);
        log("future1");
        return "Hello";
    });

CompletableFuture<String> future2 = CompletableFuture
    .supplyAsync(() -> {
        sleep(2000);
        log("future2");
        return "World";
    });

CompletableFuture<String> future3 = CompletableFuture
    .supplyAsync(() -> {
        sleep(1000);
        log("future3");
        return "Java";
    });

CompletableFuture<Object> future = CompletableFuture.anyOf(future1, future2, future3);

log("result : " + future.get());

[12:32:22.200336600][ForkJoinPool.commonPool-worker-23] future3
[12:32:22.214299300][main] result : Java

allOf()

• 모든 future의 결과를 받아서 처리하는 메소드

CompletableFuture<String> future1 = CompletableFuture
    .supplyAsync(() -> {
        sleep(3000);
        log("future1");
        return "Hello";
    });

CompletableFuture<String> future2 = CompletableFuture
    .supplyAsync(() -> {
        sleep(2000);
        log("future2");
        return "World";
    });

CompletableFuture<String> future3 = CompletableFuture
    .supplyAsync(() -> {
        sleep(1000);
        log("future3");
        return "Java";
    });

CompletableFuture<Void> future = CompletableFuture.allOf(future1, future2, future3);

String result = Stream.of(future1, future2, future3)
    .map(CompletableFuture::join)
    .collect(Collectors.joining(", "));

log("get() : " + future.get());
log("future1.isDone() : " + future1.isDone());
log("future2.isDone() : " + future2.isDone());
log("future3.isDone() : " + future3.isDone());
log("result : " + result);

[12:37:57.561345][ForkJoinPool.commonPool-worker-23] future3
[12:37:58.556364200][ForkJoinPool.commonPool-worker-5] future2
[12:37:59.562620500][ForkJoinPool.commonPool-worker-19] future1
[12:37:59.564619][main] get() : null
[12:37:59.566400100][main] future1.isDone() : true
[12:37:59.566400100][main] future2.isDone() : true
[12:37:59.566400100][main] future3.isDone() : true
[12:37:59.567397300][main] result : Hello, World, Java

참고

• https://codechacha.com/ko/java-completable-future/