[Java 8] CompletableFuture ①

CompletableFuture

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

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

• CompletableFuture는 Future과 CompletionStage을 구현했다.

• Completable 이란 이름이 붙은 이유는 외부에서 Complete을 시킬 수 있기 때문이다.

  • 몇 초 이내에 응답이 안 온다면 기본 값을 반환하도록 할 수 있다.

• CompletableFuture만을 가지고 비동기 작업을 실행할 수 있다.

Future로는 하기 어렵던 작업들

• Future를 외부에서 완료 시킬 수 없다.

  • 취소하거나, get()에 타임아웃을 설정할 수는 있다.

• 블로킹 코드( get() )를 사용하지 않고서는 작업이 끝났을 때 콜백을 실행할 수 없다.

• 여러 Future를 조합할 수 없다.

  • ex) Event 정보 가져온 다음 Event에 참석하는 회원 목록 가져오기

• 예외 처리용 API를 제공하지 않는다.

---

비동기로 작업 실행하기

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

  • 원하는 Executor(쓰레드풀)를 사용해서 실행할 수도 있다, 기본은 ForkJoinPool.commonPool()

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

        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName());
        });

        future.get();

    }
    
	// ForkJoinPool.commonPool-worker-19    

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

  • 원하는 Executor(쓰레드풀)를 사용해서 실행할 수도 있다, 기본은 ForkJoinPool.commonPool()

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

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());

            return "Hong";
        });

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

    }
    
    // ForkJoinPool.commonPool-worker-19
	// Hong

---

콜백 제공하기

• CompletableFuture를 사용하면 get 호출 전에 Callback을 정의할 수 있다.

thenApply(Function)

• 리턴 값을 받아서 다른 값으로 바꾸는 콜백

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

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());

            return "Hong";
        }).thenApply(s -> {
            System.out.println(Thread.currentThread().getName());

            return s.toUpperCase();
        });

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

    }
    
    // ForkJoinPool.commonPool-worker-19
	// main
	// HONG

thenAccept(Consumer)

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

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

        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());

            return "Hong";
        }).thenAccept(s -> {
            System.out.println(Thread.currentThread().getName());
        });

        future.get();

    }
    
    // ForkJoinPool.commonPool-worker-19
    // main

thenRun(Runnable)

• 리턴 값을 받지 않고 다른 작업을 처리하는 콜백

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

        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());

            return "Hong";
        }).thenRun(() -> {
            System.out.println(Thread.currentThread().getName());
        });

        future.get();

    }
    
    // ForkJoinPool.commonPool-worker-19
    // main

---

ForkJoinPool

ForkJoinPool은 하나의 작업 큐를 가지며 ForkJoinPool에서 관리되는 여러 쓰레드는 Parent 작업에서 분기된 Child 작업을 처리하고(Fork) 각 쓰레드에서 처리된 결과를 합쳐 Parent에게 전달해서(Join) 병렬적으로 작업을 처리하는 프레임워크다.

그렇기 때문에 쓰레드 풀을 만들지 않고도 별도의 쓰레드가 작업을 처리해 주어서 비동기 작업이 가능하다.

원한다면 직접 쓰레드 풀을 생성해서 runAsync, supplyAsync의 두 번째 인자로 넘겨주면 해당 쓰레드 풀을 CompletableFuture의 작업 쓰레드 풀로 사용할 수 있다.

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

        ExecutorService executorService = Executors.newFixedThreadPool(2);

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());

            return "Hong";
        }, executorService);

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

        executorService.shutdown();
    }
    
    // pool-1-thread-1
    // Hong

• 콜백 메서드도 별도의 쓰레드 풀에서 실행할 수 있다.

  • thenApplyAsync, thenAcceptAsync, thenRunAsync 등

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

        ExecutorService executorService = Executors.newFixedThreadPool(1);
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());

            return "Hong";
        }, executorService).thenApplyAsync(s -> {
            System.out.println(Thread.currentThread().getName());

            return s.toUpperCase();
        }, executorService);

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

        executorService.shutdown();
    }
    
    // pool-1-thread-1
    // pool-1-thread-1
    // HONG