CompletableFuture API

Future만 가지고는 비동기적인 작업을 이어서 작업하는 것이 어려웠다 -> 콜백을 줄 수가 없었으므로

CompletableFuture의 메소드를 이용하여 다양한 작업 수행

CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {
     System.out.println("hello " + Thread.currentThread().getName());
     return "hello";
});

CompletableFuture<String> world = CompletableFuture.supplyAsync(() -> {
    System.out.println("world " + Thread.currentThread().getName());
    return "world";
});

private static CompletableFuture<String> getWorld(String massage) {
    return CompletableFuture.supplyAsync(() -> {
        System.out.println("world " + Thread.currentThread().getName());
        return massage + " world";
    });
}

thenCompose()

• 두 작업이 서로 이어서 실행하도록 조합

CompletableFuture<String> future = hello.thenCompose(CompletableFuture2Main::getWorld);
System.out.println(future.get());

//출력
world ForkJoinPool.commonPool-worker-5
hello world

thenCombine()

• 두 작업을 독립적으로 실행하고 둘 다 종료했을 때 콜백 실행

//서로의 결과와 관계없이 결과를 처리하는 경우 1 _ thenCombine - 주어진 모든 결과들에 대해서 처리
CompletableFuture<String> future1 = hello.thenCombine(world, (h, w) -> {
    return h + " " + w;
});
System.out.println(future1.get());
//출력
hello world

allOf()

• 여러 작업을 모두 실행하고 모든 작업 결과에 콜백 실행

//서로의 결과와 관계없이 결과를 처리하는 경우 2
//allOf - 모든 결과가 나오고 난후 해당 결과들에 대해 처리
List<CompletableFuture<String>> futures = Arrays.asList(hello, world);

CompletableFuture[] futureArray = futures.toArray(new CompletableFuture[futures.size()]);

CompletableFuture<List<String>> futuresList = CompletableFuture.allOf(futureArray)
        .thenApply(v -> futures.stream() //get() 대신 join()을 사용 (try-catch를 피하기위해)
        .map(CompletableFuture::join)
        .collect(Collectors.toList()));

futuresList.get().forEach(System.out::println);
//출력
hello
world

allAny()

• 여러 작업중 가장 빨리 끝난 하나의 작업 결과에 콜백 실행

//anyOf - 먼저 실행되서 오는 첫번째 결과에 대해서만 받아서 처리
CompletableFuture.anyOf(hello, world).thenAccept(System.out::println);
//output
hello

예외 처리

execeptionally(Function)

boolean throwError = true;

CompletableFuture<String> Hello = CompletableFuture.supplyAsync(() -> {
    if (throwError)
        throw new IllegalArgumentException();
    System.out.println("Hello "+ Thread.currentThread().getName());
    return "Hello";
}).exceptionally(ex -> {
    System.out.println(ex);
    return "ERROR!";
});

System.out.println(Hello.get());
//출력
java.util.concurrent.CompletionException: java.lang.IllegalArgumentException
ERROR!

handle(BiFunction)

boolean throwError = true;

CompletableFuture<String> error = CompletableFuture.supplyAsync(() -> {
    if (throwError)
        throw new IllegalArgumentException();

    System.out.println("no error "+ Thread.currentThread().getName());
    return "no error";
}).handle((result, ex) -> {
    if (ex != null) {
        System.out.println(ex);
        return "ERROR";
    }
    return result;
});

System.out.println(error.get());
//출력
java.util.concurrent.CompletionException: java.lang.IllegalArgumentException
ERROR

참고

ForkJoinPool
CompletableFuture
*flow, forkjoin