[JAVA8] CompletableFuture

인프런 강의 "더 자바, JAVA8"(백기선님)의 강의를 듣고 정리한 글 입니다.
JAVA8에 추가된 핵심 기능들을 이해하기 쉽게 설명해 주시니 한번씩 들어보시는 것을 추천드립니다.

"더 자바, JAVA8 바로가기"

---

자바에서 비동기 (Asynchronous) 프로그래밍을 가능케하는 인터페이스.

• Future를 사용해서도 어느정도 가능했지만 하기 힘든 일들이 많았다.

Future로는 하기 어렵던 작업들

• Future를 외부에서 완료 시킬 수 없다. 취소하거나, get()에 타임아웃을 설정할 수는 있다.
• 블로킹 코드(get())를 사용하지 않고서는 작업이 끝났을 때 콜백을 실행할 수 없다.
• 여러 Future를 조합할 수 없다. 예) Event 정보를 가져온 다음 Event에 참여하는 회원 목록 가져오기
• 예외 처리용 API를 제공하지 않는다.

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(4);
        Future<String> future = executorService.submit(() -> "hello");

        // todo

        future.get();
        // future 에서 가져온 값을 가지고 무언가를 하는 작업은 future.get() 이후에 나와야 한다.

    }
}

CompletableFuture

• Implements Future
• Implements CompletionStage

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = new CompletableFuture<>();
        future.complete("jay"); // 명시적으로 값을 주는 방법

        String result = future.get();
        System.out.println("result = " + result); // result = jay
        
        CompletableFuture<String> future2 = CompletableFuture.completedFuture("jay");
        System.out.println(future2.get()); // jay
    }
}

비동기로 작업 실행하기

• 리턴값이 없는 경우 : runAsync()

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            System.out.println("Thread : " + Thread.currentThread().getName());
        });

        future.get();
    }
}

• 리턴값이 있는 경우 : supplyAsync()

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("Thread : " + Thread.currentThread().getName());
            return "hi";
        });

        future.get();
    }
}

• 원하는 Executor(쓰레드풀)을 사용하여 실행할 수 있다. (기본은 ForkJoinPool.commonPool())

콜백 제공하기

• thenApply(Function): 리턴값을 받아서 다른 값으로 바꾸는 콜백

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("Thread : " + Thread.currentThread().getName());
            return "hi";
        }).thenApply(String::toUpperCase);

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

• thenAccept(Consumer) : 리턴값을 또 다른 작업을 처리하는 콜백 (리턴없이)

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("Thread : " + Thread.currentThread().getName());
            return "hi";
        }).thenAccept((s) -> {
            System.out.println();
        });

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

• thenRun(Runnable) : 리턴값 받고 다른 작업을 처리하는 콜백
• 콜백 자체를 또다른 쓰레드에서 실행할 수 있다.

조합하기

• thenCompose() : 두 작업이 서로 이어서 실행하도록 조합
  (연관관계가 있는 경우)

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {
            System.out.println("hello" + Thread.currentThread().getName());
            return "hello";
        });

        CompletableFuture<String> future = hello.thenCompose(App::world);

        String s = future.get();
        System.out.println("s = " + s); // s = helloworld
    }

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

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

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        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";
        });

        CompletableFuture<String> future = hello.thenCombine(world, (h, w) -> {
            return h + w;
        });

        System.out.println(future.get()); // helloworld
    }
}

• allOf(): 여러 작얼을 모두 실행하고 모든 작업 결과에 콜백 실행
• anyOf() : 여러 작업 중에 가장 빨리 끝난 하나의 결과에 콜백 실행

예외처리

• exeptionally(Function)

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {

            if (true) {
                throw new IllegalArgumentException();
            }
            System.out.println("hello" + Thread.currentThread().getName());
            return "hello";
        }).exceptionally(ex -> {
            return "error";
        });

        System.out.println("hello = " + hello.get()); // hello = error
    }
}

• handle(BiFunction)

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {

            if (true) {
                throw new IllegalArgumentException();
            }
            System.out.println("hello" + Thread.currentThread().getName());
            return "hello";
        }).handle((result, ex) -> { // 정상동작, 에러 동작 핸들링
           if (ex != null) {
               System.out.println("error");
               return "error";
           }
            return result;
        });

        System.out.println("hello = " + hello.get()); // hello = error
    }
}