[Java8 비동기] CompletableFuture

예제는 모두 Github(https://github.com/cobiyu/CompletableFuture)에 있으니 참고 바랍니다.

문제가 되는 상황

Micro Service Architecture를 구성하게되면 여러 API로 나뉘어 지는 만큼 API끼리 서로를 호출하는 상황이 많이 발생하게 됩니다.

그리고 자연 스럽게 아래 예제와 같은 문제가 발생하게 됩니다.

public class SyncTest {
    Logger logger = LoggerFactory.getLogger(this.getClass());

    public Integer goodsPriceApi() throws InterruptedException {
        logger.info("goodsPriceApi start");
        TimeUnit.SECONDS.sleep(5);
        logger.info("goodsPriceApi success");

        return 30000;
    }

    public String userNameApi() throws InterruptedException {
        logger.info("userNameApi start");
        TimeUnit.SECONDS.sleep(5);
        logger.info("userNameApi success");

        return "user1's name";
    }

    @Test
    public void orderApi() throws InterruptedException, ExecutionException {
        Integer goodsPrice = goodsPriceApi();
        String userName = userNameApi();

        logger.info("username : " + userName + ", goodsPrice : " + goodsPrice);
    }
}

 

위 코드는 주문 API(OrderApi)를 실행하는데 상품 API(GoodsApi)와 회원 API(UserApi)를 사용하며
각 API는 모두 5초의 응답시간이 있는  상황입니다.

위 로직은 아래와 그림 같이 동작하게 됩니다.

각 API가 5초가 걸린다면 총 10초가 걸리게 된다.

실제 실행된 LOG

 

결국 주문 API는 총 10초가 걸리게되고 이는 여러 API를 호출하는 MSA에서는 더욱 문제가 될것입니다.

Java8에서는 CompletableFuture을 이용하여 비동기 프로그래밍으로 위 문제를 해결 할 수 있습니다.

 

CompletableFuture

public class CompletableFutureTest {
    Logger logger = LoggerFactory.getLogger(this.getClass());

    public CompletableFuture<Integer> goodsPriceApi() {
        return CompletableFuture.supplyAsync(() -> {
            try {
                logger.info("goodsPriceApi start");
                TimeUnit.SECONDS.sleep(5);
                logger.info("goodsPriceApi success");
                return 30000;
            } catch (InterruptedException e) {
                e.printStackTrace();
                return null;
            }
        });
    }

    public CompletableFuture<String> userNameApi() {
        return CompletableFuture.supplyAsync(() -> {
            try {
                logger.info("userNameApi start");
                TimeUnit.SECONDS.sleep(5);
                logger.info("userNameApi success");
                return "user1's name";
            } catch (InterruptedException e) {
                e.printStackTrace();
                return null;
            }
        });
    }

    @Test
    public void orderApi() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> goodsApiFuture = goodsPriceApi();
        CompletableFuture<String> userApiFuture = userNameApi();

        Integer goodsPrice = goodsApiFuture.get();
        String userName = userApiFuture.get();

        logger.info("username : " + userName + ", goodsPrice : " + goodsPrice);
    }
}

각 API를 비동기로 실행하게 되면 총 5초가 소요된다

상품 API가 실행되고 기다리는것이 아닌 비동기로 회원 API 실행

CompletableFuture를 이용해서 별도의 Thread에 작업을 지시한 후 완료될때까지 다른 작업을 동시에 실행 할 수 있습니다.

그리고 각 CompletableFuture에 get()을 실행하게 되면 블록킹이 일어나서 완료될때까지 대기하게 됩니다. 

 

 

thenCompose

위 예제로 모든것이 해결가능할 것 같지만 바로 곧 지나지 않아 풀리지 않는 상황이 바로 발생합니다.

1. 상품 API에서 가져온 가격정보를 할인 API를 통해 할인가격을 가져와야 한다.
2. 회원 API로 회원 정보를 가져와야 한다.

위 상황은 총 3개의 API를 사용하게 되는데 모든 API를 비동기로 실행하기에는 할인 API는 상품 API의 결과를 기다려야하는 상황
즉, 동기로 동작해야하는 상황입니다. 그림으로 표현하면 아래와 같습니다.

 

위 상황을 해결하기 위해서 또 3개의 API를 동기로 돌리면 응답시간은 또 다시 늘어나게 됩니다.

이런 경우를 위하여 CompletableFuture는 thenCompose라는 기능을 제공합니다.

public class ThenComposeTest {
    Logger logger = LoggerFactory.getLogger(this.getClass());

    public CompletableFuture<Integer> goodsPriceApi() {
        return CompletableFuture.supplyAsync(() -> {
            try {
                logger.info("goodsPriceApi start");
                TimeUnit.SECONDS.sleep(5);
                logger.info("goodsPriceApi success");
                return 30000;
            } catch (InterruptedException e) {
                e.printStackTrace();
                return null;
            }
        });
    }

    public CompletableFuture<Double> discountApi(Integer goodsPrice) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                logger.info("discountApi start");
                TimeUnit.SECONDS.sleep(5);
                logger.info("discountApi success");
                if(goodsPrice>=30000){
                    return goodsPrice - (goodsPrice * 0.4);
                } else{
                    return goodsPrice - (goodsPrice * 0.2);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                return null;
            }
        });
    }

    public CompletableFuture<String> userNameApi() {
        return CompletableFuture.supplyAsync(() -> {
            try {
                logger.info("userNameApi start");
                TimeUnit.SECONDS.sleep(5);
                logger.info("userNameApi success");
                return "user1's name";
            } catch (InterruptedException e) {
                e.printStackTrace();
                return null;
            }
        });
    }

    @Test
    public void orderApi() throws ExecutionException, InterruptedException {
        CompletableFuture<Double> goodsAndDiscountFuture = goodsPriceApi().thenCompose(this::discountApi);
        CompletableFuture<String> userApiFuture = userNameApi();

        String userName = userApiFuture.get();
        Double goodsDiscountPrice = goodsAndDiscountFuture.get();

        logger.info("username : " + userName + ", goodsDiscountPrice : " + goodsDiscountPrice);
    }
}

 

위 코드의 로직은 아래와 같습니다.

1. 상품 API에서 가져온 가격을 할인API에 전달
2. 할인 API는 전달된 가격이 30,000이상이면 40%할인 / 이외에는 20%할인가격 return
3. 회원 API에서 회원 정보 GET

1번과 2번은 동기로 동작하며 3번은 또 따로 비동기로 실행되는 로직입니다.

아래 실행된 콘솔을 보면 위 로직대로 동작한 것을 확인할 수 있습니다.