[JAVA]비동기 작업 파이프라인 만들기(CompletableFutrue 파이프라인)

2021.09.07 - [프로그래밍 노트/JAVA] - 병렬 스트림(parallel) vs CompletableFuture non-blocking 코드 만들기

추가된 클래스 설명

상점에서 제공한 문자열 파싱하여 Quote 클래스로 캡슐화 수행

public class Quote {

  private final String shopName;
  private final double price;
  private final Discount.Code discountCode;

  public Quote(String shopName, double price, Discount.Code discountCode) {
    this.shopName = shopName;
    this.price = price;
    this.discountCode = discountCode;
  }

  public static Quote parse(String s) {
    String[] split = s.split(":");
    String shopName = split[0];
    double price = Double.parseDouble(split[1]);
    Discount.Code discountCode = Discount.Code.valueOf(split[2]);
    return new Quote(shopName, price, discountCode);
  }

  public String getShopName() {
    return shopName;
  }

  public double getPrice() {
    return price;
  }

  public Discount.Code getDiscountCode() {
    return discountCode;
  }

}

Quote 객체를 인수로 받아 할인된 가격 문자열을 반환한다.(applyDiscount - delay 1초)

public class Discount {

  public enum Code {
    NONE(0), SILVER(5), GOLD(10), PLATINUM(15), DIAMOND(20);

    private final int percentage;

    Code(int percentage) {
      this.percentage = percentage;
    }
  }

  public static String applyDiscount(Quote quote) {
    return quote.getShopName() + " price is " + Discount.apply(quote.getPrice(), quote.getDiscountCode());
  }

  private static double apply(double price, Code code) {
    delay(); // Discount 서비스 응답 지연을 흉내낸다.
    return format(price * (100 - code.percentage) / 100);
  }

}

Discount 서비스를 이용하여 가격을 찾는 기본적인 findPrices를 구현해보면 아래와 같다.

public List<String> findPricesSequential(String product) {
  return shops.stream()
      .map(shop -> shop.getPrice(product)) // 각 상점에서 할인 전 가격 얻기
      .map(Quote::parse) // 반환된 문자열을 Quote객체로 반환
      .map(Discount::applyDiscount) // Discount 서비스를 이용해서 할인 적용
      .collect(Collectors.toList());
}

성능상 거리가 먼것을 알 수 있다. 상점이 5개라고 가정하면
getPrice → 각 상점마다 1초 씩 총 5초 소요
applyDiscount → 각 상점마다 1초 씩 총 5초 소요
총 10초가 소요되는 것을 알 수 있다.

[BestPrice price is 110.93, LetsSaveBig price is 135.58, MyFavoriteShop price is 192.72, BuyItAll price is 184.74, ShopEasy price is 167.28]
sequential done in 10107 msecs

CompletableFutrue를 이용해서 CPU 사용을 극대화 해보자

동기 작업과 비동기 작업 조합하기

public List<String> findPrices(String product) {
        List<CompletableFuture<String>> priceFutures = 
                shops.stream()
        .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor))
        .map(future -> future.thenApply(Quote::parse))
        .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor)));

        return priceFutures.stream()
                        .map(CompletableFuture::join)
                        .coolect(toList());
}

 

가격 정보 얻기(첫번째 map)

supplyAsync에 람다를 전달해서 비동기적으로 상점에서 정보를 조회한다. 첫 번째 변환의 결과는 Stream<CompletableFuture<String>>이다.

Quote 파싱하기(두번째 map)

파싱 동작은 I/O가 없으므로 원하는 즉시 지연 없이 동작을 수행한다. 생성된 CompletableFuture에 thenApply 메서드를 호출한 다음에 문자열 Quote 인스턴스로 변환하는 Function을 전달한다.

thenApply 메서드는 CompletableFuture가 끝날 때까지 블록하지 않는다는 점을 주의해야 함. 즉, CompletableFuture가 동작을 완전히 완료한 다음에 thenApply 메서드로 전달된 람다표현식을 적용할 수 있다. CompletableFuture<String> → CompletableFuture<Quote> (CompletableFuture의 결과물로 무엇을 할지 지정하는 것과 같음)

CompletableFuture를 조합해서 할인된 가격 계산하기(세번째 map)

두 가지 CompletableFuture로 이루어진 연쇄적으로 수행되는 두 개의 비동기 동작을 만들 수 있다.

1. 상점에서 가격 정보를 얻어 와서 Quote로 변환하기
2. 변환된 Quote를 Discount 서비스로 전달해서 할인된 최종가격 획득하기

자바 8의 CompletableFuture API는 이와 같이 두 비동기 연산을 파이프라인으로 만들 수 있도록 thenCompose 메서드를 제공한다. thenCompose 메서드는 첫 번째 연산의 결과를 두 번째 연산으로 전달한다. 즉, 첫 번째 CompletableFuture에 thenCompose 메서드를 호출하고 Function에 넘겨주는 식으로 두 CompletableFuture를 조합할 수 있다. 두 번째 CompletableFuture는 첫 번째 CompletableFuture의 결과를 계산의 입력으로 사용한다.

세 개의 map 연산 결과 스트림의 요소를 리스트로 수집하면 List<CompletableFuture<String>> 가 된다.
마지막으로 CompletableFuture가 완료되기 기다렸다가 join 으로 값을 추출할 수 있다.

5개의 상점에도 비동기로 수행되므로 1초씩 걸리는 것을 볼 수 있다.

(상점의 수가 쓰레드 풀의 쓰레드 갯수를 넘어가버리면 1초가 넘는 것을 기억해야한다.)
getPrice → 1초
applyDiscount → 1초
총 2초의 시간이 소요되는 것을 알 수 있다.

[BestPrice price is 204.78, LetsSaveBig price is 190.85, MyFavoriteShop price is 128.92, BuyItAll price is 140.31, ShopEasy price is 166.1]
All shops have now responded in 2011 msecs

thenCompose 메서드를 사용하였는데, thenCompose 메서드도 Async로 끝나는 버전이 있다.

Async로 끝나지 않는 메서드 - 이전 작업을 수행한 스레드와 같은 스레드에서 작업을 실행함

Async로 끝나는 메서드 - 다음 작업이 다른 스레드에서 실행되도록 스레드 풀로 작업을 제출한다.

출처 : 모던 자바 인 액션