[토킹테일] 대량 요청 처리를 위한 Spring Webflux 도입기 (1)

Choi Wontak·2025년 4월 25일
비바이빙빠르게_개발하기

비바이빙

목록 보기
6/8
post-thumbnail

비바이빙 - 빠르게 개발하기

AI 시대에 맞추어 2주에 한 프로덕트를 만들어내는, 작지만 빠른 개발을 지향하는 프로젝트입니다.

로그 폭탄이 발생한다면? (문제 인식 단계)

이전에 짠 기획을 정리하자면,

  1. 로그 자동 불러오기 기능
    URI와 토큰, 가져오는 주기, 필터링할 문자열 입력
    -> 해당 주기만큼 요청
    -> 응답 문자열에서 필터링 값 존재 확인
    -> 존재하면 해당 유저에게 알림
    가져오는 주기는 10분으로 고정, 5분은 유료 구독
    등록해둔 서버는 토킹테일 앱으로 버튼을 누르면 로그 리스트를 확인 가능 (여태껏 불러온 로그를 보여줄지, 눌렀을 때 보낸 요청으로 가져온 값만 보여줄지는 고민)

  2. 로그 알림 기능 (유료 구독 서비스)
    서버에서 발급한 토큰과 서버로 로그를 전송하면 사용자 알림으로 전송 (알림서버 개념)
    서버에서 발급한 토큰의 payload와 요청을 보내온 서버의 origin이 동일해야함.
    반복되는 불가능한 접근은 차단

1번은 Schedule을 이용하면 되기 때문에 예상 가능하지만 동기 처리를 하면 너무 느릴 것 같고, 2번은 한 서버에서 발생하는 많은 양의 로그를 처리하려면 과부하가 걸릴 것으로 생각했다.

100명의 사용자 요청이 동시에 들어오면 → 스레드 100개 생성
그 중 90개는 외부 서버에 HTTP 요청하고 대기 상태
이 대기 시간 동안 CPU는 아무 것도 하지 않지만 스레드는 메모리를 차지하고 block
EC2 프리티어를 사용하기 때문에 스레드 수가 제한될 것이다.

무엇이 필요할까 (기획 단계)

문제점

외부로 보내는 API 요청이 대량으로 발생하고, 내부로 들어오는 API 요청도 대량 발생할 것으로 예상되는 기획, 효율적으로 처리할 방법은 없을까?

해결방안
최근에 유튜브 알고리즘에 떠 시청하게 된 영상이 있었다.

[10분 테코톡] 🐅호돌의 Spring Webflux

해당 영상으로 Webflux를 처음 접하였는데 비동기/논블로킹 방식이라는 점이 문제점에 대한 해결 방식으로 작용할 수 있을 것 같아 조금 더 알아보았다.

Spring Webflux
The original web framework included in the Spring Framework, Spring Web MVC, was purpose-built for the Servlet API and Servlet containers. The reactive-stack web framework, Spring WebFlux, was added later in version 5.0. It is fully non-blocking, supports Reactive Streams back pressure, and runs on such servers as Netty, Undertow, and Servlet containers.
https://docs.spring.io/spring-framework/reference/web/webflux.html

요약하자면, Webflux는 스프링 5.0 이후 버전에 추가된 리액티브 기술의 웹 프레임워크이며 Netty와 같은 서버 위에서 동작한다.

요약해도 어려운 말들이 너무 많다. 하나씩 파헤쳐보자.

리액티브 프로그래밍

Wikipidea
In computing, reactive programming is a declarative programming paradigm concerned with data streams and the propagation of change. With this paradigm, it is possible to express static (e.g., arrays) or dynamic (e.g., event emitters) data streams with ease, and also communicate that an inferred dependency within the associated execution model exists, which facilitates the automatic propagation of the changed data flow.
https://en.wikipedia.org/wiki/Reactive_programming

데이터 스트림과 변화의 전파와 관련된 선언형 프로그래밍 패러다임이다. 정적과 동적 데이터 스트림을 쉽게 표현할 수 있고 실행 모델에서의 추론된 의존성이 존재한다는 것을 전달할 수 있다. 이를 통해 데이터 흐름이 변경될 때 그 변경 사항이 자동으로 전파되는 것이 가능해진다.

데이터 스트림의 표현과 변화를 쉽게 관리할 수 있는 패러다임인 것 같다.
변화를 감지한다..? 리액트 같은 건가 하고 GPT에게 물어보니 아니랜다.

  • 리액트: "값이 바뀌었네? → 다시 렌더해야지!" (하지만 수동으로 setState 또는 useState 같은 걸로 알려줘야 함)
  • 반응형 프로그래밍: "값이 바뀌면? → 그 값을 구독하고 있는 애들 전부 알아서 반응함" (진짜 자동 전파)

흠 이 부분은 코딩을 해봐야 알 것 같다.

Netty

Tomcat에선 클라이언트의 요청에 대해 스레드를 하나씩 할당하는데, 이 스레드는 할당된 서블릿이 동작을 모두 마칠 때까지 대기한다. 스레드가 모두 사용 중인 경우 새로 들어온 스레드에 대해선 Queue로 이동시켜 빈 자리가 생길 때까지 기다린다.
이 방식은 우리가 아무리 비동기로 프로그래밍을 하더라도, 작업을 동시에 받지 못하니 의미가 사라진다는 문제가 있다.

이에 반하여 Netty는 스레드가 기다릴 일이 생기면 콜백 처리해두고, 다른 작업을 처리하러 간다. 일이 끝나면 그때 콜백을 실행해 작업으로 돌아온다.

정리) Webflux는 스프링 5.0 이후 버전에 추가된 데이터 스트림의 표현과 변화에 대한 웹 프레임워크 기술이며, Netty와 같은 NIO 작업이 가능한 서버 위에서 동작한다.

이제 Netty + Spring Webflux를 사용해보자!

공부 내용 정리

간단한 컨트롤러를 작성해보았다.

@RestController
public class HelloController {

    @GetMapping("/hello")
    public Mono<String> hello() {
        return Mono.just("Hello, Reactive World!");
    }
}

일단 잘 작동하는 것 같다.

@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> stream() {
    return Flux.interval(Duration.ofSeconds(1))
               .take(5)
               .map(i -> "Tick " + i);
}

이번엔 비동기 이벤트를 만들어서 SSE 방식으로 String을 전달해보았다.

헉 잘 작동했다..! 짧은 코드로 이렇게 나오는 게 너무 신기하다..

@GetMapping(value = "/jokes")
    public Mono<String> getQuote() {
        return webClient.get()
                .uri("https://api.chucknorris.io/jokes/random")
                .retrieve()
                .bodyToMono(String.class);
    }

이번엔 다른 URI로 API 요청을 보내보았다.

JSON 그대로 넘어와 좀 더럽지만 어쨌든 요청이 잘 간다.
WebFlux는 RestTemplate 대신 WebClient 사용한다는 것을 알아두어야겠다.

@GetMapping("/combined")
public Mono<String> getCombined() {
    Mono<String> name = Mono.just("WebFlux");
    Mono<String> message = Mono.just("is awesome");

    return Mono.zip(name, message)
               .map(tuple -> tuple.getT1() + " " + tuple.getT2());
}

// result : WebFlux is awesome

zip은 두 모노를 합친다.
map으로 값들을 조회해 사용할 수 있다.

이제 기본적인 개념(?) 정돈 익힌 듯하다.

좀더 심화해보고 싶고 SSE가 신기하니 채팅 기능을 간단히 만들어볼까 한다.

간단한 채팅 기능 만들기

단일 채팅방에서 멀티캐스트로 메시지를 전달하는 방식을 만들어보겠다.

public class ChatMessage {
    private String sender;
    private String content;
    private LocalDateTime timestamp = LocalDateTime.now();
}

우선 메시지를 담아줄 모델을 작성해준다.

@Service
public class ChatService {

    private final Sinks.Many<ChatMessage> sink;
    private final Flux<ChatMessage> chatFlux;

    public ChatService() {
        this.sink = Sinks.many().multicast().onBackpressureBuffer();
        this.chatFlux = sink.asFlux().share();
    }

    public void sendMessage(ChatMessage message) {
        sink.tryEmitNext(message);
    }

    public Flux<ChatMessage> receiveMessages() {
        return chatFlux;
    }
}

이제 서비스 레이어에 채팅 로직을 작성하였다. 얘가 제일 중요하니까 조금 자세히 기록해보겠다.

Sink는 메시지를 푸시(push)하는 역할이고,
Many는 여러 값을 푸시할 수 있다는 뜻이다.(다수의 ChatMessage를 여러 subscriber에게 보내는 역할)

Sinks.many().multicast().onBackpressureBuffer();

multicast : 여러 클라이언트의 구독이 가능
onBackpressureBuffer: 소비자가 느릴 경우 메시지를 버퍼에 임시 저장함 (유실 방지)

sink.asFlux().share();

asFlux()는 sink를 읽을 수 있는 Flux로 변환
share()는 여러 구독자에게 동일한 Flux 스트림을 공유하게 한다.
(한 번 발행한 메시지를 모든 클라이언트가 동시에 받는다)

public Flux<ChatMessage> receiveMessages() {
    return chatFlux;
}

Flux를 반환함으로써 클라이언트가 이를 구독하는 스트리밍을 구현한다.

public void sendMessage(ChatMessage message) {
    sink.tryEmitNext(message);
}

sink에 메시지를 전송한다. 내부적으로는 Flux로 메시지를 보낸다고 한다.

@RestController
@RequiredArgsConstructor
public class ChatController {

    private final ChatService chatService;

    @PostMapping("/chat")
    public Mono<Void> sendMessage(@RequestBody ChatMessage message) {
        chatService.sendMessage(message);
        return Mono.empty();
    }

    @GetMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ChatMessage> streamMessages() {
        return chatService.receiveMessages();
    }
}

컨트롤러 로직은 위와 같다. 채팅을 받는 streamMessages 부분이 Flux를 반환해 연결을 유지하고, TEXT_EVENT_STREAM_VALUE를 이용해 받은 메시지를 SSE 방식으로 전달하고 있다.

HTML에서 작성된 부분 중 받는 부분의 js만 보면,

const eventSource = new EventSource("/chat/stream");
    eventSource.onmessage = (e) => {
        const msg = JSON.parse(e.data);
        const li = document.createElement("li");
        li.textContent = `[${msg.sender}] ${msg.content} (${msg.timestamp})`;
        document.getElementById("messages").appendChild(li);
    };

EventSource를 이용해 연결을 유지하고 있으며, onmessage로 콜백을 실행한다.

요런 화면을 두 개 띄워두고 채팅을 보내면..

보낸 내용을 확인할 수 있다.

테스트맨이 보낸 메시지도 잘 도착한 것을 확인할 수 있다.

스프링 MVC라면 SseEmitter를 이용해 직접 상태를 관리했어야 한다.
또한 스레드를 하나씩 점유했을테니 클라이언트가 늘어날수록 서버에 부담은 심해졌을 것이다.

이제 리액티브 프로그래밍에서 변화를 감지한다는 것이 무엇인지 좀 감이 올 것도 같다.

연습도 끝났고 얼른 실전에 적용해보고 싶다!

후기

이렇게 짧으면서도 효율적인 코드를 작성할 수 있다니... 리액티브 프로그래밍 진짜 매력적인 것 같다!!

profile
Choi Wontak
백엔드 주니어 주니어 개발자
이전 포스트

[토킹테일] 로그 확인이 너무 귀찮다

다음 포스트

[토킹테일] Spring WebFlux 채팅 예제로 연습하기 (1-1)

0개의 댓글