자바 8에 추가된 기술 [8] CompletableFuture

자바 8에는 다양한 기술들을 살펴보려고 하는데, 그 여덟 번째는 CompletableFuture이다.

그 전에 이 글들을 먼저 살펴보고 오자.

2022.02.07 - [Computer Science/JAVA] - java에서 멀티쓰레드 [1] Thread와 Runnable

java에서 멀티쓰레드 [1] Thread와 Runnable

2022.02.07 - [분류 전체보기] - java에서 멀티쓰레드 [2] Executors와 ExecutorService

java에서 멀티쓰레드 [2] Executors와 ExecutorService

2022.02.07 - [Computer Science/JAVA] - java에서 멀티쓰레드 [3] Callable과 Future

java에서 멀티쓰레드 [3] Callable과 Future

등장 배경

Future만으로는 비동기 프로그래밍이 제한적이었다.

Future로 처리하기 어렵던 작업들은 다음과 같다.

• 예외 처리용 API를 제공하지 않는다.
• 여러 Future를 조합할 수 없다.
• 블로킹 코드(get())를 사용하지 않고서는 작업이 끝났을 때 콜백을 실행할 수 없다.

import java.util.concurrent.*;

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        Future<String> future = executorService.submit(() -> "hello");

        // get()이 블록킹 콜이므로 최대한 get을 호출하기 전 많은 일을 해내면 되긴 함
        future.get();
        executorService.shutdown();
    }
}

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

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

CompletableFuture는 Future와 CompletionStage 인터페이스를 구현한다.

Completable이 붙은 이유는 외부에서 종료(Complete)시킬 수 있기 때문이다.

CompletableFuture를 사용하면 명시적으로 Executors를 사용할 필요가 없다.

CompletableFuture 생성 방법

import java.util.concurrent.*;

public class App {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture future= new CompletableFuture();
        // CompletableFuture future= CompletableFuture.completedFuture("eric"); // 정적 팩터리 메서드
        future.complete("eric");

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

비동기로 작업 실행하기

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

import java.util.concurrent.*;

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

        CompletableFuture<Void> future = CompletableFuture.runAsync(() ->
                System.out.println("eric: " + Thread.currentThread().getName()));

        future.get();

    }
}

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

import java.util.concurrent.*;

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

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

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

    }
}

// 실행결과
eric: ForkJoinPool.commonPool-worker-1
eric

runAsync()와 supplyAsync()는 Future와 사용할 때와 거의 같다.

콜백 제공하기

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

import java.util.concurrent.*;

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

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("eric1: " + Thread.currentThread().getName());
            return "eric";
        }).thenApply((str) -> {
            System.out.println("eric2: " + Thread.currentThread().getName());
            return str.toUpperCase();
        });

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

    }
}

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

import java.util.concurrent.*;

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

        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("eric1: " + Thread.currentThread().getName());
            return "eric";
        }).thenAccept((str) -> {
            System.out.println("eric2: " + Thread.currentThread().getName());
            System.out.println(str.toUpperCase());
        });

        future.get();
    }
}

• thenRun(Runnable) : 리턴값에 상관없이 다른 작업을 처리하는 콜백

import java.util.concurrent.*;

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

        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("eric1: " + Thread.currentThread().getName());
            return "eric";
        }).thenRun(() -> {
            System.out.println("eric2: " + Thread.currentThread().getName());

        });

        future.get();
    }
}

조합하기

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

import java.util.concurrent.*;

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

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

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

    private static CompletableFuture<String> getHello(String message) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("hello: " + Thread.currentThread().getName());
            return message + " hello";
        });
    }
}

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

import java.util.concurrent.*;

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

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

        CompletableFuture<String> future = hello.thenCombine(eric, (h, e) -> h + " " + e);

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

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