[Spring] 스프링 웹플럭스 Spring WebFlux

CastleSilver·2022년 11월 2일

Spring Framework 5.0버전 부터 추가되었다. Spring WebFlux는 논블로킹으로 동작하며, Reactive Streams back pressure를 지원하고 Netty, Undertow, Servlet 3.1+ 컨테이너와 같은 서버에서 실행된다.

개요

Spring WebFlux가 만들어진 이유

적은 수의 스레드로 동시 처리하고 더 적은 하드웨어 리소스를 가지고 확장 가능한 non-blocking web statck이 필요했기 때문이다.
함수형 프로그래밍을 위해서이다. Java 8에서는 람다 표현식이 추가되면서 자바에서도 함수형 API를 작성할 수 있게 되었다. 이는 논블로킹 어플리케이션을 만들 때 용이하며 continuation-style API로 비동기 로직을 선언적으로 작성할 수 있다.

Reactive Streams 란?

논블로킹(Non-blocking) 백 프레셔(back pressure)를 이용한 스트리밍 방식의 비동기 데이터 처리의 표준

스트리밍 처리

전통적인 데이터 처리 방식은 데이터 처리 요청이 오면 페이로드(payload) $_{전송되는 데이터}$ 를 모두 애플리케이션의 메모리에 저장한 다음 처리를 해야 한다. 추가되는 데이터도 모두 DB에서 조회하여 메모리에 적재해야 한다. 이 방식의 문제점은 1. 필요한 데이터의 크기가 메모리 용량보다 클 때 발생하는 out of memory ERROR 2. 순간적으로 많은 요청이 몰리면 발생하는 다량의 GC(Garbage Collection)으로 인한 서버의 응답 오류 발생이다.
스트림 처리 방식을 적용하면 크기가 작은 시스템 메모리로도 많은 양의 데이터를 처리 할 수 있다. 입력 데이터에 파이프 라인을 만들어 데이터가 들어오는 대로 구독, 처리, 발행까지 한 번에 연결하여 처리 가능하기에 서버는 많은 양의 데이터도 유연하게 처리할 수 있다.

비동기 방식

동기 방식은 클라이언트가 서버에 요청을 보내면 응답을 받기 전까지 블로킹(blocking)이 된다. 따라서 A에 요청을 보내면 응답이 온 후에야 B에 요청을 보낼 수 있다.
비동기 방식은 현재 스레드가 블로킹이 되지 않기에 A에 요청을 보낸 후 응답을 기다릴 필요 없이 바로 B에 요청을 보내거나 다른 일을 처리할 수 있다.
따라서 비동기 방식은 동기 방식에 비해 다음의 장점을 가진다. 1. 여러 요청을 동시에 보내기에 더 빠른 응답 속도를 보여준다. 2. 현재 스레드가 블로킹되지 않고 다른 업무를 처리할 수 있어 더 적은 수의 스레드로 더 많은 양의 요청을 처리할 수 있다.

백 프레셔

옵저버 패턴(observer pattern)에서는 푸쉬 방식으로 데이터가 전송되었다. 즉 publisher가 subscriber에게 밀어 넣는 방식으로 데이터가 전달되고 이 과정에서 subscriber의 상태는 고려되지 않았다. subscriber에 제때 처리하지 못하는 이벤트는 큐(queue)에 저장되었다.
버퍼가 다 소모되어 오버플로(overflow)가 발생할 경우 1. 고정길이 버퍼는 신규 메시지를 거절하고 거절된 메시지는 재요청된다. 재요청 과정에서 네트워크와 CPU 연산 비용이 추가로 발생한다. 2. 가변길이 버퍼는 'out of memory' 에러가 발생하면서 서버 크래시(crash)가 발생한다.
위 문제를 해결하기 위해 발행자가 데이터를 전달할 때 구독자가 필요한 만큼만 전달하는 것이 백 프레셔의 기본 원리이다.
백 프레셔는 풀 방식으로 이루어진다. 풀 방식에서는 구독자는 처리할 수 있는 만큼만 발행자에게 요청하고 발행자는 요청받은 만큼만 전달하면 된다. 즉 풀 방식에서는 전달되는 모든 데이터의 크기를 구독자가 결정한다.

Reactive Streams

데이터가 지속적으로 흐르기 위해 Netflix와 Pivotal, Lightbend의 엔지니어들이 협동하여 개발한 라이브러리

Reactive Streams API

public interface Publisher<T> {
   public void subscribe(Subscriber<? super T> s);
}
 
public interface Subscriber<T> {
   public void onSubscribe(Subscription s);
   public void onNext(T t);
   public void onError(Throwable t);
   public void onComplete();
}
 
public interface Subscription {
   public void request(long n);
   public void cancel();
}

publisher
1. 구독자를 받기 위한 subscribe API
subscriber
1. 받은 데이터를 처리하기 위한 onNext API
2. 에러를 처리하기 위한 onError API
3. 작업 완료 시 사용하는 onComplete API
4. 매개 변수로 구독을 받는 onSubscribe API
subscription
1. n개의 데이터를 요청하기 위한 request API
2. 구독을 취소하기 위한 cancel API