IT로 세상을 이롭게

<가상의 시나리오>

- Ingestion Layer에서 수백개의 병렬처리를 통해서 데이터를 생성하고 있으며, 해당 데이터에 접근할 수 있도록 파티션별로 Raw API 제공되고 있음

- Role, Scheduler, Auth 등등 여러 가지 문제 때문에 신규 API 를 만드는 데 시간이 필요함 (파티션별로 나누어진 데이터를 합쳐서 연산해야 함)

-  Application Layer에서 Raw API들을 반복 호출하여 결과값을 연산하도록 로직을 구성하도록 하며 최대한 처리속도를 끌어올리고 자원효율을 극대화하자.

<접근방법>

- 다수의 API 호출 후 결과를 조합해야 하는 경우 Web client를 활용하여 비동기 호출로 효과를 봤었다. (이전 포스트 참조)

https://icthuman.tistory.com/entry/Spring-WebClient-%EC%82%AC%EC%9A%A9%EC%8B%9C-%EC%A3%BC%EC%9D%98%EC%A0%90

https://icthuman.tistory.com/entry/Spring-WebClient-%EC%82%AC%EC%9A%A9-2-MVC-WebClient-%EA%B5%AC%EC%A1%B0

https://icthuman.tistory.com/entry/Spring-WebClient-%EC%82%AC%EC%9A%A9-3-Configuration-Timeout

- 신규 프로젝트에서는 기술스택을 Spring WebFlux 로 선정하였다. 그 이유는 다음과 같다.

a. 기본적으로 Spring, Java에 대한 이해도가 높다. 하지만 Legacy 코드는 없다.

b. 데이터에 대한 읽기 연산이 대부분이고, 특별한 보안처리나 트랜잭션 처리가 필요없다. (참조해야할만한 Dependecny 가 적다.)

c. 저장공간으로 Redis Cache를 활용한다. 즉, Reactive를 적극 활용할 수 있다.

d. 다수의 API 호출을 통해서 새로운 결과를 만들어 낸다.

즉, IO / Network의 병목구간을 최소화 한다면 자원활용을 극대화 할 수 있을 것으로 보인다.

<진행내용>

- 기존의 For loop 방식과 Async-non blocking 차이,그리고 Mono / Flux 를 살펴본다. (Spring WebFlux) 

@ReactiveRedisCacheable
public Mono<String> rawApiCall(...) throws .Exception {

Mono<String> response = webClient
                .get()
                .uri(url)
                .retrieve()
                .onStatus(HttpStatus::is4xxClientError, clientResponse -> Mono.error(new Exception(...)))
                .onStatus(HttpStatus::is5xxServerError, clientResponse -> Mono.error(new Exception(... )))
                .bodyToMono(String.class)
                .timeout(Duration.ofMillis(apiTimeout))
                .onErrorMap(ReadTimeoutException.class, e -> new Exception(...))
                .onErrorMap(WriteTimeoutException.class, e -> new Exception(...))
                .onErrorMap(TimeoutException.class, e -> new Exception(...));
                
                return response;
}

webClient를 이용해서 타 API를 호출하는 부분이다. 응답값에는 다수의 건이 포함되어 있으나 해당 데이터를 보내는 쪽에서도 병렬처리를 진행하고 있기 때문에 Collection 이나 Array 형태로 처리하는 부분을 제외하고 그냥 Raw line 형태로 제공하고 있다.

Spring MVC기반에서는 이 값을 꺼내기 위해서 결국 block하고 값에 접근하는 로직이 필요하다. 굳이 코드로 구현하자면 아마도 이렇게 만들어 질 것이다.

List<ApiResponse> ret = new ArrayList<>();
for(String value : Collection ... ){

   String contents = apiService.rawApiCall(value).block();

   String[] lines = contents.split("\n");
   for(String data : lines){
       if(StringUtils.hasText(data)){

           ApiResponse apiResponse =  mapper.readValue(data, ApiResponse.class);

           if(populationHourApiResponse .. ){
               // biz logic
				
               FinalResponse finalResponse = new FinalResponse();
               // setter
               ...
               ..
               
               ret.add(finalReponse);
           }
       }
   }
}

이 코드에는 여러가지 문제점이 있는데

- block()을 수행하게 되면 비동기 넌블러킹 처리의 여러 장점이 사라진다.

- 오히려 더 적은 수의 쓰레드를 사용해야 하는 구조특성상  block이 생기면 더 병목이 발생하는 경우도 있다.

- return 에 얼만큼의 데이터가 담길지 모르게 된다.

- API Call 이후 biz logic의 수행시간이 길어질 수록 전체 응답시간은 더욱 길어진다.

해당 내용을 block없이 처리하도록 Flux를 최대한 활용하여 작성해보았다.

public Flux<FinalResponse> getDataByConditionLevel1{

    List<Mono<String>> monoList = new ArrayList();
    for(String value : Collections ...)){
        monoList.add( apiService.rawApiCall(value) );
    }

    return 
        Flux.merge(monoList)
                .flatMap(s -> Flux.fromIterable(Arrays.asList(s.split("\n"))))
                .filter(s -> StringUtils.hasText(s))
                .map(data -> {
                    try {
                        return mapper.readValue(data, PopulationApiResponse.class);
                    } catch (JsonProcessingException e) {
                        log.error(e.getLocalizedMessage());
                    }
                    return new ApiResponse();
                })                                                                      
                .filter(aApiResponse -> ... biz logic)     
                .map(apiResponse ->
                     new FinalResponse(...)
                );
  }

주요하게 바뀐부분을 살펴보면 다음과 같다.

1. API응답의 결과를 block해서 기다리지 않고 Mono를 모아서 Flux 로 변환한다.

Mono는 0..1건의 데이터, Flux는 0..N건의 데이터를 처리하도록 되어있다.

즉 개별 Mono를 대기하여 처리하는 것이 아니라 하나의 Flux로 모아서 단일 Stream처럼 처리할 수 있다,.

2.  값이 아니라 행위를 넘겨준다.

Spring WebFlux에서는 기본적으로 Controller - Service - Dao 등의 Layer간 이동을 할때 Mono / Flux 를 넘겨준다.

즉, 어떠한 값을 보내는 것이 아니라 Mono / Flux로 구성된 Publisher를 전달해주면 subscribe를 통해서 실제 데이터가 발생될 때 우리가 정의한 Action을 수행하는 형태가 된다고 이해하면 될듯 하다. (Hot / Cold 방식의 차이가 있는데 일단 Skip하도록 한다.)

위의 로직은 각 개별 데이터 간의 연산이나 관계가 없기 때문에 비교적 쉽게 변경할 수 있었다.

하지만 해당 데이터를 다시 조합하거나 Grouping 하거나 하는 경우가 있다면 약간 더 복잡해질 수 있기 때문에 고민이 필요하며 각 비지니스 케이스에 적합한 단위와 연산으로 재설계를 해주는 것이 좋다. ( -> 필수다 !)

예를 들어서 rawApiCall에 필요한 인자값이 yyyyMMdd hh:mm:ss 형태의 timeStamp라면 특정기간 내 시간대별 결과를 얻기 위해서는 다음과 같이 Call을 하고 조합해야 한다.

즉 수행해야 하는 액션은 다음과 같다.

- Flux를 응답받는 메소드를 다시 감싸서

- 응답결과를 적절하게 Biz Logic에 따라서 처리한 뒤

- aggreation 을 통하여 새로운 응답을 만들어 낸다. (e.g 그룹별 개수, 합계, 평균 등등)

코드로 작성해보면 이러한 형태가 될텐데

public Flux<NewResponse> getDataByConditionLevel2{	

    List<Flux<NewResponse>> ret = new ArrayList();
        for( ; ; ){
            ...
            // Biz Logic...
            ...
            
            Flux<NewResponse> flux = getDataByConditionLevel1( ...  )
                     .groupBy(apiSummary -> apiSummary.getKey() )
                     .flatMap(groupedFlux -> groupedFlux.reduce( (arg1, arg2) -> ApiSummary.add(arg1, arg2) )
                                                        .map(apiSummary -> NewResponse.valueOf( ...+ groupedFlux.key(), apiSummary ))
                                                      );

            ret.add(flux);
        }
        return Flux.merge(ret);

 위의 코드에서 살펴볼 부분은 세 가지이다.

- groupBy : getDataByConditionLevel1 메소드에서 받아온 결과를 Key단위로 Grouping을 수행한다.

  이때 수행결과로는 GroupedFlux가 리턴되는 데 이는 중첩된 데이터 형태로 flatMap 을 통해서 작업하는 것이 수월하다.

- reduce : groupBy 로 분류된 데이터들을 key 단위로 reduce 하게 되는데 ( 자주 보게되는 wordCount sample과 유사하다).

  Java내에는 Integer , Double등의 타입에서 ::sum 메소드를 제공하고 있지만 우리가 직접 작성한 Class 에 대해서는 연산메소드를 정의해주는 것이 필요하다. 위 예제에서는 ApiSummary.add(arg1, arg2) 이다.

 최종 객체변환의 편의성을 위해서 NewResponse.valueOf 메소드도 정의해서 사용하였다.

- Mono/Flux간 변환

getDataByConditionLevel1 메소드에서 살펴본것 처럼 여러 개의 Mono는 하나의 Flux로 변환이 가능하다.

또한 Flux에 대한 reduce 연산은 Mono로 변환이 된다.

그리고 여러 개의 Flux 를 합쳐서 하나의 Flux로 변환하는 것도 가능하다.

 순서보장이 필요한지, 병렬처리가 필요한지 등 여러가지 요건을 고려하여 적절한 연산자를 사용하도록 한다.

<정리>

- 처음에는 blocking 로직을 벡엔드에서 가지고 있는 것이 적합하지 않아서 FrontEnd에서 해당 API들을 호출하여 결과값을 연산하는 형태로 접근했었다. (Promise all)

- 일주일치의 데이터를 기반으로 결과값을 생성하기 위해서는 총 24 * 7 = 168 회 API 호출이 필요했고, 프론트에서 처리시간은 최악의 경우 15초를 넘어가는 케이스가 발생하였다.

- Spring Web Flux를 활용하여 Backend에서 처리하도록 개선하였으며 또한 Raw API Call을 수행하는 메소드에 별도로 개발한 Cache Aspect를 적용하였다.

그 이유는 Spring Cache Manager에서 async/non-blocking에 대한 표준 구현체가 없다보니 직접 CacheMono/Flux와  ReactiveRedisTemplate등을 사용하여 값을 처리하도록 구현하였다.

이에 대한 내용은 다음 포스트에서 좀 더 자세히 다루도록 하겠다.

<결과>

- 최초 호출시 약 4~5초 정도 수행시간이 소요되며, 각 Raw API 캐시 이후에는 약 1초 정도 걸리는 것을 확인할 수 있었다.

  Network 처리에 가장 많은 시간이 소요되기 때문에 사실 개별 API Call만 캐시해도 성능이 대폭 향상된다.

- 하지만 아직 몇 가지 더 살펴보고 싶은 욕심이 있는데.. 

 a. Mono / Flux 레벨에서의 캐시

 b. Raw API뿐만 아니라 최종 API에 대한 값 캐시 

 (각 Raw API 응답값이 변하기도 하고, 워낙 대상이 많다보니 캐시대상을 늘릴 경우 저장공간에 대한 우려가 있다.)

 c. Reactor에서의 병렬처리

 (Schedulers, parallel 등)

<참조>

Reactor에 대한 내용이 잘 정리되어 있다.

https://godekdls.github.io/Reactor%20Core/reactorcorefeatures/

https://icthuman.tistory.com/entry/Reactive-Programming-1-%EA%B4%80%EB%A0%A8-%EA%B0%9C%EB%85%90%EC%A0%95%EB%A6%AC

spring.codec.max-in-memory-size=20MB

ExchangeStrategies exchangeStrategies = ExchangeStrategies.builder()
                .codecs(clientCodecConfigurer -> clientCodecConfigurer.defaultCodecs().maxInMemorySize(3* 1024 * 1024))
                .build();

        return WebClient.builder()
                .exchangeStrategies(exchangeStrategies)
                .baseUrl( )
                .clientConnector(new ReactorClientHttpConnector(httpClient))
                .build();

HttpClient httpClient = HttpClient.create()
                .tcpConfiguration(
                        client -> client.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, -A- ) //miliseconds
                                .doOnConnected(
                                        conn -> conn.addHandlerLast(new ReadTimeoutHandler( -B- , TimeUnit.MILLISECONDS))
                                                .addHandlerLast(new WriteTimeoutHandler( -C- , TimeUnit.MILLISECONDS))
                                )
                );


        return WebClient.builder()
                .exchangeStrategies(exchangeStrategies)
                .baseUrl(webClientProperties.getBaseUri())
                .clientConnector(new ReactorClientHttpConnector(httpClient))
                .build();

Mono<String> response = webClient
                .get()
                .uri(" ")
                .retrieve()
                .onStatus(HttpStatus::is4xxClientError, clientResponse -> Mono.error(new Exception())
                .onStatus(HttpStatus::is5xxServerError, clientResponse -> Mono.error(new Exception())
                .bodyToMono(String.class)
                .timeout(Duration.ofMillis( -D- ))
                .onErrorMap(ReadTimeoutException.class, e -> ))
                .onErrorMap(WriteTimeoutException.class, e -> ))

A. Connection Timeout

- Client / Server 간의 Connection을 맺기 위해 소요되는 시간을 의미한다.

- Server에서 새로 Connection을 맺을 자원의 여유가 없다면 발생할 수 있다.

- HTTP Connection에 대한 내용이기 때문에 keep-alive 옵션역시 사용가능하다.

B. Read Timeout

 - 데이터를 읽기 위해서 기다리는 시간을 의미한다.

 - 내가 호출한 API 가 응답을 주지 못하면 발생할 수 있다.

 - 너무 길다면? 적절히 설정해주지 않으면 응답을 받을때까지 계속 대기하게 되고, 자원이 고갈되는 현상이 발생한다.

 - 너무 짧다면? 요청을 받은쪽에서는 처리가 되었으나, 응답을 기다리던 쪽에서는 Timeout이 발생하게 되어서 불일치 상태가 발생한다. 

C. Write Timeout

 - 데이터를 쓰기 위해서 기다리는 시간을 의미한다.

 - 주어진 시간동안 Write Operation을 완료하지 못하면 발생할 수 있다.

 - 즉, Connection 연결 후 데이터를 보내기 시작하였으나 해당시간보다 길어지게 되면 중단된다.

D. reactor timeout

 - Reactive Stream은 Publisher, Subscriber, Subscription 을 통해서 비동기 / 넌블러킹 / back pressure 처리하는 개념이다.

 - 우리가 다루는 Spring WebFlux는 reactive stream의 구현체로 reactor를 사용하고 있으며 Mono / Flux가 Publisher의 구현체이다.

 - 따라서 Exception , Retry등을 처리할때도 기존 방식 대신 reactive stream의 기능을 활용해주는 것이 장점을 충분히 살릴 수 있는 방법이라고 생각한다.

 - Spring WebFlux에서는 WebClient의 호출결과를 받았을때 결과 Body를 Mono로 감싸주어 데이터를 전달하는 형태가 되는데, 해당 시간동안 데이터를 전달하지 못하게 되면 timeout 이 발생하게 된다.

E. Transaction Timeout과 비교 (개인적인 생각 , 틀릴 수 있음)

 - 우리가 일반적으로 DB transaction timeout을 설명할 때 Transaction Timeout > Statement Timeout > Socket Timeout 로 각 구간을 나누어서 설명하고 상위 Layer( ? 포장레이어?) 에서는 하위 Layer보다 크거나 같게 설정하는 것이 일반적이다.

 - Web 호출 역시 비슷하게 살펴본다면 Publisher Timeout > Read/Write Timeout > Connection Timeout 정도로 비슷하게 정리해 볼 수 있지 않을까 생각했다.

<정리>

- MSA구조에서 각 API의 응답시간과 사이즈는 적절하게 설정해야 한다.

- 특히, 각 레벨에서의 적절한 수준의 timeout 설정은 필수이다. 

- 너무 짧으면 많은 오류가 발생하게 되고, 이에 따른 side-effect (데이터 불일치, 로직의 복잡도 증가) 가 생기게 되며

- 너무 길거나 무제한으로 설정하게 되면 리소스 자원의 낭비가 발생한다.

<현상>

- Async API Call 후 응답을 제대로 처리하지 못하는 현상이 있습니다.

- 그 여파로 내부적으로 AtomicInteger를 이용하여 호출Count를 처리하는 로직이 있는데 해당 로직이 수행되지 않아서 버그가 발생하고 있었습니다.

<원인>

contents = webClient
                .post()
                .uri(multiCountApiPath)
                .accept(MediaType.APPLICATION_JSON_UTF8)
                .contentType(MediaType.APPLICATION_JSON_UTF8)
                .header("Authorization","Bearer " + token)
                .body(BodyInserters.fromObject(inputs))
                .retrieve()
                .bodyToMono(String.class)
                .timeout(Duration.ofMillis(200000))
                .onErrorReturn(null)
                .flux()
                .toStream()
                .findFirst()
                .orElse(null);

        try {
            ...
        } catch (IOException e) {
            ...
        }
        callCount.decrementAndGet();
    
        return CompletableFuture.completedFuture(ret);

- 원래의 의도는 API Call 오류가 발생하였을 경우 null 로 처리하여 빈값을 가져가도록 하는 것이었습니다.

- 그러나 실제로 테스트해보면 try 이후 구문이 수행되지 않고 있습니다.

- API Call에서 오류가 발생했을 경우 null 처리를 잘못하여 flux() 이하 로직이 수행되지 않고 있습니다.

<수정사항>

contents = webClient
                .post()
                .uri(multiCountApiPath)
                .accept(MediaType.APPLICATION_JSON_UTF8)
                .contentType(MediaType.APPLICATION_JSON_UTF8)
                .header("Authorization","Bearer " + token)
                .body(BodyInserters.fromObject(inputs))
                .retrieve()
                .bodyToMono(String.class)
                .timeout(Duration.ofMillis(200000))
                .onErrorReturn("")
                .flux()
                .toStream()
                .findFirst()
                .orElse("");

        try {
            ...
        } catch (IOException e) {
            ...
        }
        callCount.decrementAndGet();
    
        return CompletableFuture.completedFuture(ret);

     

- 위와 같이 bodyToMono에서 우리가 사용하고자 했던 타입에 맞는 값(e.g String "" )으로 처리해주면 onErrorReturn 이후 로직이 정상적으로 수행되는 것을 확인할 수 있습니다.

<추가로 확인해야 할 사항>

- timeout 이 발생했을 때 특정 로직을 수행하도록 handler 가 가능하다면 decrement 로직을 그쪽으로 옮길 수 있을지 검토가 필요합니다.

- onErrorReturn 이후 로직이 Spring내부에서 어떻게 동작하는지 상세한 확인이 필요합니다.