Spring WebFlux 이해하기 - Reactive Streams

Spring WebFlux 의 작동방식을 이해하기 위해 단계별로 진행하는 포스팅입니다.
이 글에서는 Reactive Streams 에 대해 설명합니다.

 

Reactive Programming

 

리액티브 프로그래밍은 빠른 반응을 하고자 하는 시스템입니다.

클라이언트의 요청에 즉각적으로 응답함으로써 지연 시간을 최소화하고

이런 빠른 응답을 바탕으로 유지보수와 확장이 용이한 시스템을 구축하고자 합니다.

 

리액티브 선언문에 따르면, 리액티브 프로그래밍은 아래 4가지를 만족해야 합니다.

• 응답성 : 시스템이 가능한한 즉각적으로 응답
• 탄력성 : 시스템이 장애에 직면하더라도 응답성을 유지하는 것
• 유연성 : 시스템의 작업량이 변화하더라도 응답성을 유지하는 것
• 메세지 구동 : 비동기 메세지 전달에 의존하여 구성요소 간 느슨한 결합, 격리, 위치투명성을 보장

위의 설명에 의하면 Non-Blocking IO, Async, HA 와 같은 키워드들이 떠오릅니다.

그리고 이는 현대의 MSA 아키텍쳐에 아주 적합해보입니다.

Scale-Out 으로 빠른 반응성을 추구하는 MSA 아키텍쳐는 위와 같은 구성이 필요하기 때문입니다.

Reactive Streams 는 이러한 패러다임을 지키기위한 표준 spec 입니다.

 

Reactive Streams

Reactive Streams is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure.

 

Reactive Streams 공식 홈페이지에서는 위와 같이 설명하고 있습니다.

데이터 스트림을 Non-Blocking, Backpressure 로 비동기적인 방식을 제공한다고 합니다.

 

Stream, Non-blocking, Async

 

Java NIO 의 작동 원리

 

 

스트림 / 비동기 처리는 위의 글에서 이해하실 수 있습니다.

payload 를 메모리에 적재해서 전송하는 방식이 아닌 chunk 단위로 나눠 전송하는 것을 의미합니다.

또한 비동기 방식을 지원하므로 데이터를 전송할 때 현재 스레드가 대기하지 않고 다른 일을 할 수도 있습니다.

이는 HTTP2 의 전송 방식과 비슷합니다.

 

Backpressure

Backpressure 는 받는 쪽의 입장을 고려한 process 라고 생각할 수 있습니다.

데이터 처리를 하며 받는 쪽에서 시간이 지연됐음에도 불구하고 데이터를 지속해서 보낸다면 

데이터 유실 혹은 시스템의 장애가 발생할 것을 우려해 데이터를 소비하는 쪽에서 얼마만큼의 데이터를 요청하는 것을 의미합니다.

이는 push/pull 방식 중 pull 방식에 기인합니다. 말 그대로 data 를 pull 하는 것입니다.

 

 

Reactive Streams 의 구성 요소

• Publisher
  • 데이터를 발생하는 Producer
• Subscriber
  • 데이터를 소비하는 consumer
  • Subscriber 는 Publisher 를 구독
• Subscription
  • Publisher 와 Subscriber 사이의 구독 관계를 나타냄
  • Subscriber 는 Subscription 을 통해 Publisher 에게 데이터를 요청하거나, 구독을 취소할 수 있음
• Processor
  • Publisher 와 Subscriber 역할을 모두 수행하며 데이터를 가공하거나 변환

단순히 Message Broker 에 전송하는 Kafka 와 같은 시스템에서 보던 Publihser/Subscriber 를 생각한다면 

Subscription 의 존재는 의아할 수 있습니다.

간략히 설명하자면 Subscription 은 Message Broker 와 비슷한 역할을 합니다.

코드로 살펴보겠습니다.

 

public class PubSub {
    public static void main(String[] args) {
        // 1 - 5 까지 생성
        Iterator<Integer> iterator = List.of(1, 2, 3, 4, 5).iterator();

        // Publisher 생성, Data Stream 생성
        Flow.Publisher<Object> publisher = new Flow.Publisher<>() {
            @Override
            public void subscribe(Flow.Subscriber<? super Object> subscriber) {
                // 구독을 시작하면 Scubscription 을 subscriber 에게 전달
                subscriber.onSubscribe(new Flow.Subscription() {
                    @Override
                    public void request(long n) {
                        // subscriber 가 데이터를 요청하면, 요청한 개수만큼 데이터를 전달
                        iterator.forEachRemaining(subscriber::onNext);
                        subscriber.onComplete();
                    }

                    @Override
                    public void cancel() {
                        // Subscriber 가 구독을 취소
                    }
                });
            }
        };

        Flow.Subscriber<> subscriber = new Flow.Subscriber<>() {
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                // Publisher 로부터 Subscription 객체를 받으면 데이터 요청을 시작
                System.out.println("PubSub.onSubscribe");
                subscription.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Object item) {
                // 데이터를 받으면 호출되는 메서드
                System.out.println("item = " + item);
            }

            @Override
            public void onError(Throwable throwable) {
                // 데이터 전송 중 오류 시
                System.out.println("PubSub.onError");
            }

            @Override
            public void onComplete() {
                // 모든 데이터를 받으면 호출됨
                System.out.println("PubSub.onComplete");
            }
        };

        // publisher 가 subscriber 에 구독 요청을 하면 subscriber.onSubscribe 가 호출되며 구독이 시작
        publisher.subscribe(subscriber);
    }

}

 

Publisher 는 Data Stream 을 전송하며 Subscriber 의 구독을 받게 되고,

Subscriber 는 Publisher 에게 Subscription 이라는 객체를 받아 상호 작용을 하게됩니다.

Subscription 은 request 로 구독의 요청, cancel 로 구독의 취소를 하고

subscriber 가 Subscription.request 이후 onNext / onError / onComplete 를 수행하게 됩니다.

 

이때 마지막 줄에서 '데이터를 제공하는 publisher 가 왜 subscribe 를 호출하지?' 라는 생각이 드실 수 있는데요.

이는 신문사(publisher) 와 구독자 (subscriber) 의 관계로 생각하시면 편합니다.

구독자가 신문사에 신문 구독을 신청한다는 것을 코드로 표현한 것입니다.

 

따라서 데이터를 생성하는 주체 / 데이터를 소비하는 주체를 따로 두어 상호작용을 통해 

비동기 Async 시스템을 만족합니다. 관련된 사항은 Reactor 에서 자세히 살펴보겠습니다.

 

Reactive Streams 의 구현체로는 RxJava, Reactor, Akka Streams, Java 9 Flow 가 있습니다.

Spring WebFlux 는 Reactor 를 사용하므로 다음 글은 Reactor 의 사용법에 대해서 작성하도록 하겠습니다.

 

 

끝!

 

참고

https://www.reactivemanifesto.org/ko

리액티브 선언문

https://www.reactive-streams.org/

https://www.reactive-streams.org/