리액티브 프로그래밍으로 이벤트 처리 속도, 스레드 사용량 개선

처리량 개선이 필요하다.

회사가 성장하며 기기 이벤트가 크게 늘었다. 8개월 전까지만 해도 초당 1500건이었던 기기 이벤트가, 이제는 3500건을 넘는다. 한 달 동안의 이벤트 수 변화량만 보더라도, 그 수위가 점점 높아지는 것을 볼 수 있다. 기존 인프라와 로직으로는 늘어난 유입량을 못 따라가고, 곧 지연과 유실로 이어진다.

 

한 달 동안의 이벤트 수 매트릭

 

이벤트 컨슘부터 Ack까지의 흐름 안에는 다양한 로직들이 있다. 그래서 '어떤 구간이 문제다!'라기보다는, 주변 영향을 최소화하면서도 효과적인 개선 방향을 고민한다. 물론 어떤 경우에는 단순히 인프라만 늘리면 해결되는 경우도 많다. 프로세스를 더 띄우고, 스레드 수를 늘리는 게 제일 빠를 수도 있겠지만, 최대한 주어진 자원 안에서 개선하는 방법을 고민하는 게 요즘 우리 팀의 방향이다. 

 

지난번엔 캐시 적용이었고, 이번에는 리액티브 프로그래밍이다. 반응형 라이브러리를 사용한 논-블록킹으로 I/O 대기를 피하여, DB 적재를 처리하는 로직의 처리량을 개선하거나 외부 API 호출에 사용하는 스레드 개수를 줄인 경험을 소개한다.

 

리액티브 프로그래밍

리액티브 프로그래밍은 데이터 변화나 이벤트 처리에 반응하며 로직을 전개하는 꼴을 말한다. 예를 들어 DB 저장이나 외부 API 호출을 하는 상황에서 요청 후 처리 완료 이벤트가 도착했을 때, 이에 반응해 그 결과를 처리하는 식으로 로직을 진행한다. 이를 비동기로 한다면, 요청 후 이후 흐름은 이어가고, 처리 완료 후 결과를 처리하는 식으로 I/O 대기를 피할 수 있게 되는 것이다.

 

 

처음 이 '대기 없이 이어감'을 들었을 때는, 새로운 스레드를 사용하는 것 외에는 방법이 떠오르지 않았다. 개선을 위해 사용했던 방식은, 그보다는 커널 단에서의 멀티플렉싱에 더 답이 있었다. I/O 대기에 스레드를 사용하지 않기에 보다 비용 효율적이다. Reactive MongoDB나 WebClient는 java.nio와 Netty와 같은 라이브러리 위에서 동작한다. 그리고 이들은 커널 단의 epoll(리눅스), kqueue(Mac), IOCP(Window)와 같은 멀티플렉싱 I/O 함수를 기반으로 한다. 멀티플렉싱이란 FD를 하나의 스레드에서 감시하고, 이벤트 처리가 완료된 FD를 프로그램 단으로 올려주는 기법을 말한다. 

 

반응형 프로그래밍과 논 블록킹을 구분해야겠다.
반응형은 이벤트 변화에 반응하여 로직을 전개해 나가는 코딩 패러다임이지, 논-블록킹을 사용해야 하는 것은 아니다.
단지 자바/Spring 진영에서 사용하는 반응형 프로그래밍이 논-블록킹과 함께 쓰이는 경우가 많아 익숙해진 것은 아닌지 생각해본다.

 

처리 순서 주의

처리 완료를 기다리지 않은 흐름이기에, 반대로 순서 처리에 주의해야 한다. 네트워크 상태, 워커 스레드 풀 상태에 따라 먼저 수신한 메시지가 그 후에 처리한 메시지보다 늦게 처리될 수도 있게 된다. 예를 들어 카프카를 통해 메시지를 수신하고 이를 DB에 저장하는 로직을 가정하자. 카프카 수신부터 DB 적재 요청, 카프카 ACK까지의 한 흐름을 처리하는 데는 빠르지만, 네트워크 이슈로 먼저 DB 적재가 요청던 이벤트 A보다, 그 후에 요청된 B가 더 빠르게 적재될 수 있는 것이다.

 

 

워커 스레드 수 주의 

이런 반응형 프로그래밍과 논 블록킹 조합이, '스레드를 전혀 사용하지 않는다'는 것은 아니다. Spring 생태계에서 Netty를 이용한 논-블록킹은, I/O 요청이 후 완료까지 스레드 대기가 불필요하다는 것이지, 완료 후 I/O 이벤트가 발생했을 때부터의 후처리는 마찬가지로 스레드가 필요하다. 이때 Netty의 EventLoopGroup 스레드 풀이 사용된다.

 

그렇기에 이 스레드 풀을 사용하는 후처리가 길어지면, 스레드 풀이 마르는 현상을 똑같이 발생할 수 있다. 특히 Netty의 기본 스레드 풀 사이즈는 코어 수 * 2로, 톰캣의 기본 수나 보통 애플리케이션에서 설정하는 스레드 풀 수보다 작기 때문에, 이를 고려하지 않은 리액티브 프로그래밍은 오히려 병목을 만들 수 있다.

 

Github - Netty, EventLoopGroup의 스레드 수 결정 방법

public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {
    private static final int DEFAULT_EVENT_LOOP_THREADS;
    static {
        DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
                "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
    }
}

 

Spring MVC와 함께 사용할 때

이런 리액티브 프로그래밍을 서버 애플리케이션 전체에 적용하면, 웹 요청도 논-블로킹 방식으로 처리할 수 있게 된다. Spring 생태계에서는 WebFlux가 대표적이다. 기존의 Spring MVC는 톰캣을 기반으로 요청마다 하나의 스레드를 할당하지만, Netty 기반의 WebFlux는 이벤트 루프를 이용해 스레드 대기를 최소화하고, 동시에 더 많은 요청을 처리할 수 있다.

 

그렇다고 WebFlux가 언제나 정답은 아닌 것 같다. 리액티브 프로그래밍은 익숙한 블록킹 방식보다 코드 작성이 어렵고, 기존 방식에서의 전환도 쉽지 않다. 또 그렇다고 트래픽이 많지 않은 서비스가 아니라면 기존 블록킹 방식과의 처리량 차이도 미미하다. 요청 수가 많아질수록 WebFlux와 MVC의 격차가 커진다.

그렇기에 기존 Spring MVC로 요청당 스레드 사용을 유지하면서, Reactive MongoDB나 WebClient와 같은 리액티브 라이브러리를 사용하는 것도 방법이다. 하지만 이 경우에는 전체 요청 처리 흐름에서 완전한 논-블록킹 효과를 얻을 수 없다. 아래와 같은 컨트롤러라고 하면, WebFlux에서는 Mono가 구독되고 요청 스레드가 대기 없이 먼저 반환된 후에, 외부 요청에 대한 응답 이벤트가 도착하면, 이를 이벤트 루프의 스레드가 후처리 하여 응답하는 꼴이다.

 

반면, Spring MVC에서는 컨트롤러가 Mono <String> 같은 리액티브 타입을 반환해도, 컨트롤러가 리턴하는 시점까지 톰캣 스레드는 반환되지 않는다. 즉, 사용자 요청 처리 흐름 안에서 이벤트 루프로 논-블록킹으로 처리한다고 해도, 결국 사용자 요청 처리 스레드는 그 논-블록킹 처리 완료를 대기해야 하는 것이다.

 

@GetMapping("/example")
public Mono<String> callApi() {
    return webClient.get()
      .uri("http://external-api.com")
      .retrieve()
      .bodyToMono(String.class);
}

 

개선 로직 1. DB 저장 서비스, 처리량 47% 증가

이벤트 파이프라인에서 전달되는 이벤트는 여러 BE 서비스에서 본인의 역할에 맞게 처리된다. 그중 History 서비스는 기기의 상태 이벤트를 저장하거나, 그 내역을 조회할 수 있도록 한다. DB는 MongoDB를 사용하고 있고, 글을 쓰는 시점에서, 해당 서비스로 초당 2500개의 메시지가 전달되고 있다. 

 

기존에는 DB에 Insert를 전달하고 대기하였다가, 저장을 마치면 다음 메시지를 수신하는 흐름이었다. Reactive MongoDB를 도입하여 DB에 Insert를 전달하고 대기 없이 바로 다음 메시지를 수신하는 꼴로 로직을 변경했다. 기기 상태 이력이야 조회하는 시점에서 정렬하니 저장 순서가 완벽히 보장되지 않아도 되었고, 이벤트 수신부터 Ack 전달까지의 사이클에 드는 시간을 최소화하는 것이 중요한 로직이었기에, 리액티브 도입에 적합할 것이라고 생각했다.

 

 

적용 결과는 매우 효과적이었다. 메시지 처리 사이클이 매우 짧아졌고, DB에 데이터 적재에도 지연이 없었다. MongoDB 쪽의 매트릭도 이상 없었다. 아래는 개선 전후의 처리량을 확인하기 위한 로그이다. 명확한 계산을 위해 테스트를 위한 컨슈머 스레드는 단일로 구성했다.

 

우선 개선 전의 처리량이다. 위는 카프카에 유입되는 이벤트 개수이고, 아래는 이를 수신하여 블록킹으로 DB에 적재한 개수이다. 유입 속도를 따라가지 못하고, 대략 80%의 처리량을 갖는 것을 볼 수 있다. 한 스레드에서 초당 약 1700개의 이벤트를 적재할 수 있었다.

 

Blocking - Kafka produce

 

Blocking - Kafka consume

 

이번엔 개선 후의 처리량이다. 위는 카프카로 유입되는 이벤트 개수, 아래는 이를 수신하여 논-블록킹으로 DB에 적재한 개수를 표현한다. 당연하게도 같은 시간대 유입 수와 처리 수가 완전히 동일한 값을 갖진 못하지만, 처리량 자체는 100%로 모든 메시지를 커버하는 것을 볼 수 있다.

 

Non blocking - Kafka produce

 

Non blocking - Kafka consume

 

해당 시점의 카프카 대시보드에도 특이사항은 없다. Lag에 쌓이는, 즉 지연이 발생한 경우는 없었고, 이벤트 유입과 수신 속도가 일치했다. 논-블록킹으로 전환 전에는 여러 파티션으로 처리하던 메시지량을, 전환 후에는 단일 파티션만으로 커버할 수 있었다.

 

초당 처리량이 1,700건에서 2,500건으로 약 47% 향상된 것을 확인할 수 있었다. 처리해야 할 이벤트 수가 더 많아질 수 록 개선 효과는 더욱 커질 것으로 예상된다.

 

카프카 대시보드, Lag와 유입/수신율

 

개선 로직 2 - API  다중 호출, 단일 스레드로

기기의 상태를 조회하기 위해선 여러 Api를 호출하고 그 결과를 수집해야만 하는 제품이 있다. 이 경우 각 Api 호출을 순차적으로 처리하게 된다면 사용자 응답이 너무 늦기에, 멀티 스레드로 동시에 Api를 호출하고 그 결과를 조합한다. 지금까지는 그저 이전에 해오던 방식으로, 스레드 풀을 남발하며 응답 시간을 줄이고 있었다.

 

리액티브를 공부하고 도입하면서, 이를 스레드 풀 없이 단일 스레드로 처리할 수 있음이 떠올랐다. WebClient로 여러 요청을 논-블록킹으로 요청하고, 응답 결과만 블록킹으로 조합하면 된다면, 굳이 매 요청별로 스레드를 따로 사용하지 않을 수 있겠다는 것이었다. 

 

리액티브 방식으로 전환 후, 예상대로 스레드 사용량이 줄었는지, 요청 처리량이 낮아지진 않았는지 테스트가 필요했다. 테스트 환경을 만들고, 해당 서비스의 현재 초당 유입량만큼의 부하테스트를 진행한다. 문제가 되었던 로직에서 사용하는 스레드 풀을 분리하여, 다른 처리에 스레드 풀이 오염되는 상황을 피한다.

 

논 블록킹 전환 전, 멀티 스레드

 

논 블록킹 전환 후, 단일 스레드

 

위는 5분 동안 70개의 가상 사용자가 문제 Api에 동시 요청을 진행한 결과이다. 1초에 한번 스레드 풀의 활성 스레드 개수와 큐 개수를 집계하고 이를 출력했다. 5분을 마치고는 그 기간 동안 처리한 요청 수와 평균값을 출력한다.

 

전환 전에는 평균 287개의 활성 스레드가 필요했으나, 전환 후에는 요청 스레드 1개로만 처리되었다. 특히 지정된 스레드 풀 안에서 즉시 처리가 불가능하여 스레드 대기가 필요했던 멀티 스레드 방식에 비해, 전환 후에는 스레드 자원이 부족하여 대기하는 경우가 없었다. 그러면서도 기존 16500개의 요청 처리보다 많은 18000개를 처리할 수 있었기에, 혹여나 의심했던 요청 처리량이 정상인지에 대한 걱정도 피할 수 있었다.

 

그렇다고 WebClient로 무한히 많은 요청을 동시에 전달할 수 있는 것은 아니다.
WebClient 내부의 커넥션 풀 사이즈가 정해져 있고, 사용할 수 있는 FD, 즉 소켓 수도 한정적이다.

 

정리

작업 : 리액티브 프로그래밍과 논 블록킹 라이브러리 도입

처리 :

   - Reactive MongoDB 도입하여, DB 처리 대기 시간 제거
   - Api를 멀티스레드로 다중 호출하던 로직에서, WebClient를 사용한 단일 스레드 로직으로 전환

결과 : 

   - DB 저장 서비스의 처리량 개선, 초당 처리량이 1,700건에서 2,500건으로 약 47% 향상

   - 스레드 사용 효율 증가, 전환 전 287개의 평균 활성 스레드가 필요했던 로직을 단일 스레드로 처리