reactive programing [이론 정리]

리액티브 프로그래밍

 

리액티브 프로그래밍은 비동기적인 데이터 스트림을 다루기 위한 프로그래밍 패러다임이다. 이 패러다임은 데이터의 흐름을 중심으로 개발을 진행하며, 데이터의 변경에 반응하여 비동기적으로 처리합니다. 리액티브 프로그래밍은 확장성과 반응성이 중요한 애플리케이션에서 유용하게 활용된다.

 

리액티브 프로그램의 필요성

 

1년 치 구독료를 지불하였는데 배달이 오지 않고 1년 치 신문이 모두 준비되면 그제야 배달을 시작한다면 어떻게 될까?

실제로는 기사가 최신일 때 독자가 읽을 수 있도록 출간 후 가능한 빨리 배달된다. 또한 독자가 기사를 읽는 동안 기자는 새로운 다음 기사를 작성한다. 이 모든 것은 병행으로 진행된다. 

이처럼 애플리케이션 코드를 개발할 때는 명령형(imperative)와 리액티브(reactive 반응형) 두 가지 형태의 코드를 작성할 수 있다.

 

리액티브 프로그래밍의 특징

- 선언형 프로그래밍 방식을 사용한다.
- 지속적으로 데이터가 입력으로 들어올 수 있다.
- 데이터가 지속적으로 발생하는 것 자체를 하나의 데이터 플로우로 보고 데이터를 자동으로 전달한다.
- 데이터의 어떤 변경을 이벤트로 간주하고, 이벤트가 발생할 때마다 데이터를 계속해서 전달한다.

 

개발자 입장에서 정의

Reactive programming is programming with asynchronous data streams.

Reactive Programming의 핵심은 모든 것을 비동기적 Data의 Stream으로 간주하고,
Observer 디자인 패턴( 변화한 subject가 Observer에게 자신의 변화를 전달 )을 활용하여 이러한 비동기 이벤트를 처리하는 것이다.

 

비동기 이벤트: 프로그램에서 다양한 작업이 동시에 수행되는 중에, 어떠한 작업이 완료되는 것.

ex) 유저가 앱을 로그인, 이미지를 조회, 카메라를 작동, 이미지를  업로드 등 다양한 행위가 서버와 교신할 때마다 알게 모르게 비동기적 이벤트로써 발생하게 된다.

 

 

Data Stream

 비동기적인 Data Stream은 아래와 같은 이미지로 표현할 수 있다.

Data Stream이 값을 생성한 후 분출하는 것을 Emit 이라고한다.

 

Reactive Programming을 위해 유저의 입력에 즉각적으로 반응하려면, 프로그램이 지속적으로 값을 관찰(Observe)해야한다. 그리고 값의 변화가 일어날 때마다 특정 연산을 수행해야하는데, 이러한 디자인 패턴을 Observer 패턴이라고하며, 비동기 이벤트를 처리하는 Reactive Programming의 근간이 된다.

여기서 Observe하는 것을 Reactive Programming에서는 해당 Data Stream에 가입(subscribe)한다고도 표현한다.

이는 라디오의 주파수를 맞추거나, 관련 소식을 접하기 위해 해당 사이트의 이메일 스트림에 가입하는 것과 유사한 개념이다.

 

쉽게 설명하자면 명령형 프로그래밍(리액티브 프로그래밍의 반대 개념)에선 x=y+z라는 함수의 값은 추후에 y,z 값이 변경되어도 바뀌지 않는다.

하지만 리액티브 프로그래밍에선 y,z값이 바뀌는 Event가 발생하게 되면, 비동기적으로 Observer 패턴을 통해 변화를 알려주고, 이에 맞는 처리를 수행하게 된다.

 

 

CallBack VS Observer

 

비동기 이벤트를 처리할 때, Observer가 아닌 CallBack 패턴을 사용할 수 도 있다.

CallBack함수를 함수의 파라미터(인자)로 넘겨주는 것으로, 본 함수의 작업이 완료된 뒤, 파라미터에 들어간 CallBack함수를 호출하는 것으로 비동기 처리를 수행한다.

실제로 대부분의 Async 로직들은 CallBack으로 설계되어 있지만, 이는 굉장히 복잡한 구조가 될 가능성이 높다( CallBack Hell 문제).

 

Functional Reactive Programming

FRP는 Reactive Programming을 함수형 프로그래밍의 원리를 통해 구현하는 것이다.

 

Functional Programming(함수형 프로그래밍)

어떤 문제를 해결하는데 있어서, 이미 만들어진 함수를 활용하는 방식이다.

여기서 포인트는 함수 자체에 숨겨진 input, output가 없도록 하는 것이다.

숨겨진 input, output은 side effect라고도 부르는데, 이러한 side effect들이 많을 수록 함수를 이용할 떄 예측불허한 상황이 발생하고, 디버깅도 어려워진다.

 

ex) [1,2,3] 이라는 배열에서, 짝수의 값만 가져오는 함수를 만들 때, for문으로 탐색하여 결과를 추출하여 리턴할 수 있다.

하지만, 이러한 구현을 최대한 배제하고, filter,map,reduce등의 메소드만 사용한다면, for문 구현 없이 예측하기 쉬운 함수를 만들 수 있다(filter, map, reduce등은 이미 구현된 함수로, input, output이 명확하기에).

 

 

ReactiveX

RX라는 약어로 불린다. FRP의 원리를 통해 비동기적 이벤트를 처리하기 위해 만들어진 API이다.

즉, RX의 목적은 비동기 이벤트 계의 map, filter, reduce가 되는 것이라고도 볼 수 있다.

RX의 핵심 원리는 다음과 같다.

"RX에서는 모든 것이 Data Stream"

RX는 Stream을 Observable이라고 표현하며, 실제 라이브러리에서도 Observable객체로 선언된다.

그리고 해당 Data Stream, 즉 Observable이 언제, 무엇을 Emit하던지 Listen하기 위해 Subscribe를 수행하게 된다.

 

 

사용자 입장에서의 정의

사용자 입장에서 리액티브 프로그래밍이란, "실시간으로 반응하는" 프로그래밍을 뜻 한다.
검색창에 검색어를 입력할 때마다 자동완성이 수행되거나, instagram에서 실시간으로 내가 보고 있는 게시글의 좋아요가 증가하는 등, "실시간" 에 관한 것이라고도 설명할 수 있다.

간단하게 instagram의 예시를 통해서 "리액티브 프로그래밍" 이 어떻게 Ovserver 패턴과 비동기적 이벤트 처리 방식으로 수행되는지 알아보자.

 

상황

A유저: 피드의 게시물 조회하던 중 X게시글에 좋아요를 누르고 이어서 게시글 목록을 조회한다.
B 유저: B유저는 얼마 뒤, 피드를 조회하던 중 X게시글을 조회하게 된다.

1) A유저의 입장에서 살펴보자.

event 1: A유저가 X 게시글에 좋아요를 누른다.
event 2: A유저는 좋아요를 누르고, 이어서 게시물을 넘겨서 조회한다.

 A유저가 X라는 게시글에 좋아요를 누르게 되면, instagram앱이 서버와 교신할 때마다 알게 모르게  Observer 패턴을 통해 "A유저가 X 게시글에 좋아요 1개 추가" 라는 비동기적 이벤트를 알려주게 된다.
이는 비동기적 이벤트이기에, 당연히 A 유저는 좋아요를 누른 것과 무관하게 아무런 딜레이 없이 게시글을 계속해서 조회한다.

2) B 유저의 입장에서 살펴보자.

event 1: B유저가 모르게, instagram앱은 서버와 교신하며 "A유저가 X 게시글에 좋아요 1개 추가" 라는 비동기적 이벤트를 Observer 패턴을 통해 전달 받는다.
event 2: B유저는 피드를 계속해서 내리고, X 게시글을 지나치게된다.

백그라운드에서 Observer 패턴을 통해 "A유저가 X 게시글에 좋아요 1개 추가" 라는 비동기적 이벤트가 들어왔기에,
B 유저는계속해서 인터페이스와 상호작용하여 게시글들을 조회하는 와중에 변경된 정보를 조회할 수 있다.

 

위의 예시와 같이, 비동기처리 방식과 데이터 변경에 반응하는 패러다임(Observer 패턴 등)을 통해 "유저의 인터페이스를 방해하지 않고, 백그라운드에서 정보를 가져오는 것" 을 "리액티브 프로그래밍"이라고 칭할 수 있다. 

스프링은 Reactive Library Stack을 통해 이러한 리액티브 프로그래밍을 구현할 수 있다.
동시성 처리를 위해 async, non-blocking I/O 방식을 사용하고, 요청과 응답을 데이터 스트림으로 처리한다.
또한, 기존 Node.js에서 비동기 처리에 사용하는 CallBack 패턴이 가진 가진 콜백 지옥 문제를 해결하고, 복잡한 동시성과 에러 핸들링을 쉽게 처리할 수 있다.

 

물론 Node.js에서도 리액티브 프로그램을 통해 JavaScript에서 지원하는 라이브러리인  RXJS를 통해 리액티브 프로그래밍을 구현할 수 있고, async, await을 통해 콜백 지옥 문제를 완화할 수 있기에 절대적 장점으로 볼 수는 없다.

 

이제 스프링의 Reactive Stack에서 리액티브 프로그래밍을 위해 사용하는 리액터 라이브러리에 대해 알아보자.

 

 리액터 라이브러리

- Reactor는 RxJava 2 와 함께 Reactive Stream 의 구현체이기도 하고, Spring Framwork 5부터 리액티브 프로그래밍을 위해 지원되는 라이브러리 이다. JVM 위에서 동작하는 Non-Blocking 어플리케이션을 만들기 위해 쓰인다.

 

리액티브 프로그래밍은 비동기적이다. 따라서 동시에 여러 작업을 처리하며 높은 확장성을 가지고, back-pressure는 데이터를 소비하는 컨슈머가 처리할 수 있는 만큼의 데이터만 전달하도록 제한하여 지나치게 빠른 데이터 소스로부터 데이터 전달 과부화를 방지한다. 

 

우선 리액터를 알아보기 전에, 리액터가 구현하고 있는 Reactive Stream 에 대해 알아보자.

 

 

Reactive Stream Interface

아래 4개의 API Components로 구성되어 있다.

1. Publisher
2. Subscriber
3. Subscription
4. Processor

 

Publisher

 Publisher는 데이터를 생산한다.

하지만, 기본적으로 Subscriber가 등록되기(구독하기)전까지 아무런 일도 하지 않고, Subscriber가 등록되는 시점부터 데이터를 push 하게 된다.

또한 publisher는 하나의 Subscription 당 하나의 Subscriber 에 발행(전송)하는 데이터를 생성한다.

public interface Publisher<T> {
    void subscribe(Subscriber<? super T> subscriber);
}

 

 

 

Subscriber

Subscriber가 구독 신청되면 Publisher 로부터 이벤트를 수신할 수 있다.

앞서 말한 것과 같이,  서브젝트의 변화를 Observe하는 것을 Reactive Programming에서는 해당 Data Stream에 가입(subscribe)한다고도 표현한다.

이후 Subscriber는 Publisher에게 피드백을 주는데, 이는 혼잡을 제어할 수 있는 도구로 "모든 이벤트를 다 push하지 말고, N개만 보내줘"와 같은 요구에 해당한다.

이로써 Subscriber에 데이터가 지나치게 쌓여서 감당할 수 없는 현상을 방지한다.

이러한 피드백을 관장하는 요소가 바로 Back-Pressure 이다.

public interface Subscriber<T> {
    void onSubscribe(Subscription sub); // Subscriber가 구독할 첫 이벤트를 호출 
    void onNext(T item); // publisher가 전송하는 데이터가 Subscriber에게 전달
    void onError(Throwable ex); // 에러 발생시 호출됨
    void onComplete(); // publisher가 Subscriber 에게 작업 종료를 공지
}

 

BackPressure

리액티브 시스템에서 높은 응답성을 가장 핵심적인 개념이 BackPressure이다.

이를 통해 트래픽이 몰리더라도, 서버가 터지지 않고, Reactive하게 동작하도록 돕는다.

기존의 옵저버 패턴은 Push 방식으로, Publisher가 제한없이 데이터를 emit하는 반면, BackPressure는

Pull 방식으로, Subscriber가 Publisher에게 자신이 처리 가능한 양만 요청하는 방식이다.

 

Back-Pressure이 존재하지 않을 때

데이터의 Consumer는 자신이 감당하지 못하는 속도로 생산되는 데이터의 속도를 감당하지 못하는 문제가 발생하게 된다.

 

Back-Pressure이 존재할 때

Back-Pressure이 Producer의 생산속도를 Consumer가 따라갈 수 있도록 조절해주기에, 문제가 발생하지 않는다.

 

 

Subscription

Subscriber가 구독 신청되면 Publisher 로부터 이벤트를 수신할 수 있다.

public interface Subscription {
    void request(long n); // 전송되는 데이터 요청, n은 백프레셔 즉 데이터 항목수
    void cancel(); // 구독 취소 요청
}

 

Processor

Processor 인터페이스는 Subscriber와 Publisher 를 결합한 것이다.

public interface Processor<T, R> 
            extends Subscriber<T>, Publisher<R> {}

 

 

이제 대략적으로 리액티브 라이브러리에 대해 보았으니, 본론으로 돌아가서 Reactor에 대해 살펴보자.

 

 

Reactor

리액터란 리액티브 스트림을 구현하는 라이브러리이며 Flux 와 Mono 두가지 타입으로 스트림을 정의한다.

 

 

명령형 vs 리액티브

 

명령형 프로그래밍 예시

순차적으로 연속 되는 작업
각 작업은 한번에 하나씩 그리고 이전 작업 다음에 실행된다.
데이터는 모아서 처리되고 이전 작업이 데이터 처리를 끝낸후에 다음 작업으로 넘어간다.

String name = "shyswy";
String capitalName = name.toUpperCase();
String greeting = "Hello, " + capitalName + "!";
System.out.println(greeting);

스레드에서 한 단계씩 차례대로 실행되며, 각 단계가 완료될 때까지 다음 단계로 이동하지 못하게 실행 중인 스레드를 막는다.

 

리액티브 코드

병렬로 일련의 작업들이 진행된다.
각 작업은 부분 집합의 데이터를 처리할 수 있다.
처리가 끝난 데이터를 다음 작업에 넘겨주고 다른 부분 집합의 데이터로 계속 작업 할 수 있다.

Mono.just("shyswy")
    .map(n -> n.toUpperCase())
    .map(cn ->  "Hello, " + cn + "!")
    .subscribe(System.out::println);

일련의 작업 단계를 기술하는 것이 아니라 데이터가 전달될 파이프 라인을 구성하는 것이다. 이 파이프라인을 통해 데이터가 전달되는 동안 어떤 형태로든 변경 또는 사용될 수 있다.

 

 

Mono 와 Flux 란?

리액터는 리액티브 스트림을 구현하는 라이브러리로 Mono 와 Flux 2가지 데이터 타입으로 스트림을 정의한다.

즉, Spring webflux 를 사용하여 비동기적인 데이터 스트림의 처리를 리액터 라이브러리가 제공하는 데이터 타입인 Mono 와 Flux 로 다뤄야한다. 따라서, webflux에서는 모든 응답을 Mono, 혹은 Flux에 담아서 반환해야한다.

 

이 2가지 데이터 타입은 Stream을 정의하기 때문에, Ractor는 최소 Java8에서 동작하며, Java8의 피쳐를 잘 지원해야 사용할 수 있다.

 

Mono VS Flux

Mono: 0~1개의 결과만 처리하기위한 객체

FLux: 0~N개의 결과를 처리하기 위한 객체

 

Reactor를 사용해 일련의 스트림을 코드로 작성하다 보면 보통 여러 스트림을 하나의 결과를 모아줄 때 Mono를 쓰고, 각각의 Mono를 합쳐서 여러 개의 값을 처리하는 Flux로 표현할 수도 있다.

 

Flux<T>: Flux는 Reactive Stream의 Publisher(데이터 생산자)에 해당하는 객체이다.

 

Flux의 공식 레퍼런스를 보면, Flux가 리액티브 스트림에서 정의한 Publisher를 implements하고 있다.

 

따라서 Flux는 item들을 생산(방출)하고, Subscriber가 subscribe하기 전까지는, 아무 동작도 수행하지 않는다.

 

Flux는 onNext 이벤트가 발생 이후 0개부터 N개까지의 T타입의 원소를 생산(방출)하게 된다.

그리고 onComplete 이벤트가 발생하면 완료하고, onError 이벤트가 발생하면 에러를 발생시킵니다. 

두 이벤트는 터미널 이벤트로 흐름을 종료시킵니다. 이러한 터미널 이벤트가 발생(트리거)하지 않으면, Flux는 무한히 유지됩니다. ( 각 이벤트는 에러 발생 or 완료로 종료된다.)

 

 

 

Mono<T>

Flux와 마찬가지로 Reactive Stream의 Publisher를 Implements한다.

 

따라서 Flux는 item을 생산(방출)하고, Subscriber가 subscribe하기 전까지는, 아무 동작도 수행하지 않는다.

하지만, Mono는 Flux와 달리, 0~1 개의 아이템만 생산(방출) 가능하다.

 

이후 마찬가지로 onNext 신호를 통해 최대 하나의 아이템을 방출하고, onComplete 신호를 통해 Mono가 온전히 종료되었다는 것을 표현하고, onError를 통해 Mono의 실패를 나타낼 수 있다.

Mono와 Flux모두 Reactive Stream의 Publisher 인터페이스를 구현하고 있으며, Reactor에서 제공하는 풍부한 연산자들(operators)의 조합을 통해 스트림을 표현할 수 있다는 장점을 가지고 있다(이후의 글에서 자세히 설명 예정).

ex

- Flux에서 하나의 결과로 값을 모아주는 reduce연산자는 Mono를 리턴

- Mono에서 flatMapMany라는 연산자를 사용하면 하나의 값으로부터 여러 개의 값을 취급하는 Flux를 리턴.

그리고 Publisher인터페이스에 정의된 subscribe메서드를 호출함으로써 Mono나 Flux가 동작하도록 할 수 있다.

 

 

여기까지, 리액티브 프로그래밍의 이론적 지식에 대해 알아보았습니다.

이후의 글에선 reactor를 통한 리액티브 프로그래밍의 실습 예제를 보며 사용법에 대해 정리해보겠습니다.