[Java 8] CompletableFuture ②

CompletableFuture, 작업의 조합

• Future만으론 작업을 이어서 수행하는 것이 어려웠다.

  • Callback이 없었기에 비동기적인 작업 두 개를 연결하는 것 자체가 어려웠다.

	// day가 끝나고 dream를 불러오려면 두 번의 get 호출이 이어져야 했다.
    
	public static void main(String[] args) throws InterruptedException, ExecutionException {

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

            return "day ";
        });

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

            return "dream ";
        });

        String res1 = hello.get();
        String res2 = world.get();

        System.out.println(res1 + res2);

    }
    
    // day ForkJoinPool.commonPool-worker-19
	// dream ForkJoinPool.commonPool-worker-5
	// day dream 

---

thenCompose

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

• CompletableFuture의 thenCompose를 사용하면 연관관계가 있는 비동기 작업을 이어서 수행할 수 있다.

• thenCompose는 두 작업 간의 의존성이 필요할 때 사용된다.

  • 예를 들어 A 작업이 수행된 다음 B 작업을 수행해야 하는 상황

	public static void main(String[] args) throws InterruptedException, ExecutionException {

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

            return "day";
        }).thenCompose(s -> CompletableFuture.supplyAsync(() -> s + "dream"));

        System.out.println(daydream.get());

    }

thenCombine

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

	public static void main(String[] args) throws InterruptedException, ExecutionException {

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

            return "one";
        });

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

            return "two";
        });

        CompletableFuture<String> onetwo = hello.thenCombine(world, (o, t) -> o + t);

        System.out.println(onetwo.get());

    }

allOf

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

• allOf로 조합한 작업의 결과는 항상 void를 반환

• 만약 각 작업의 결과를 반환받고 싶다면 futures의 stream을 열어 결괏값을 List<Object>로 받을 수 있다.

	public static void main(String[] args) throws InterruptedException, ExecutionException {

        CompletableFuture<String> daydream = CompletableFuture.supplyAsync(() -> {
            return "daydream";
        });

        CompletableFuture<Void> empty = CompletableFuture.runAsync(() -> {
        });

        CompletableFuture<Integer> one = CompletableFuture.supplyAsync(() -> {
            return 1;
        });

        CompletableFuture[] futuresArray = {daydream, empty, one};
        List<CompletableFuture> futures = Arrays.asList(futuresArray);

        CompletableFuture<List<Object>> results = CompletableFuture.allOf(futuresArray).thenApply(v -> {
            return futures.stream()
                    .map(o -> o.join()) // ignore unchecked exception
                    .collect(Collectors.toList());
        });

        results.get().forEach(System.out::println);

    }

anyOf

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

	public static void main(String[] args) throws InterruptedException, ExecutionException {

        CompletableFuture<String> daydream = CompletableFuture.supplyAsync(() -> {
            return "daydream";
        });

        CompletableFuture<Integer> blog = CompletableFuture.supplyAsync(() -> {
            return 2;
        });

        CompletableFuture<Object> future = CompletableFuture.anyOf(daydream, blog);

        System.out.println(future.get());

    }

---

CompletableFuture, 예외 처리

exceptionally(Function)

• 비동기 작업에서 에러가 발생한다면 exceptionally에서 에러 타입을 받고 반환할 수 있다.

	public static void main(String[] args) throws InterruptedException, ExecutionException {

        boolean throwError = true;

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            if (throwError) throw new IllegalArgumentException();
            System.out.println(Thread.currentThread().getName());

            return "daydream";
        }).exceptionally(e -> "error!");

        System.out.println(future.get());

    }

handle(BiFunction)

• 정상적으로 실행되는 상황과 에러가 발생했을 때의 상황 모두 다룸

	public static void main(String[] args) throws InterruptedException, ExecutionException {

        boolean throwError = true;

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            if (throwError) throw new IllegalArgumentException();
            System.out.println(Thread.currentThread().getName());

            return "daydream";
        }).handle((result, ex) -> {
            if (ex != null) return "error!";

            return result;
        });

        System.out.println(future.get());

    }