[스프링 인 액션] Chapter 10 - 리액터 개요 :: 리액티브 프로그래밍 이해하기

애플리케이션 코드를 개발할 때는 명령형(imperative)과 리액티브(reactive) 두 가지 형태로

코드를 작성할 수 있다.

• 명령형 : 순차적으로 연속되는 작업이며, 각 작업은 한 번에 하나씩 그리고 이전 작업
  
  다음에 실행된다. 데이터는 모아서 처리되고 이전 작업이 데이터 처리를 끝낸 후에 다음 
  
  작업으로 넘어갈 수 있다.
  
  
• 리액티브 : 데이터 처리를 위해 일련의 작업들이 정의되지만, 이 작업들은 병렬로
  
  실행된다. 그리고 각 작업은 부분 집합의 데이터를 처리할 수 있으며, 처리가 끝난
  
  데이터를 다음 작업에 넘겨주고 다른 부분 집합의 데이터로 계속 작업할 수 있다.

리액터는 스프링 프로젝트의 일부분인 리액티브 프로그래밍 라이브러리다. 그리고 리액터는

스프링 5에서 리액티브 프로그래밍을 지원하는 데 필요한 기반이므로 먼저 리액터를 파악한 

다음에 리액티브 Controller와 Repository를 빌드하는 것이 중요하다. 일단 리액터의 사용을

시작하기 전에, 리액티브 프로그래밍의 핵심을 간단하게 알아보자.

---

리액티브 프로그래밍은 명령형 프로그래밍의 대안이 되는 패러다임이다. 명령형 프로그래밍의

한계를 해결할 수 있기 때문이다. 사실 우리가 작성하는 대부분의 코드는 명령형일 것이다.

명령형 프로그래밍의 발상은 간단하다. 한 번에 하나씩 만나는 순서대로 실행되는 명령어들로

코드를 작성하면 된다. 

명령형 프로그래밍은 낭비하는 자원이 생길 수 밖에 없다. 어떤 작업이 원격지 서버로부터 

데이터를 가져오는 것이라면, 이 작업이 완료될 때까지 아무 것도 할 수 없다. 이 작업을 

수행하는 스레드는 차단된다. 자바를 비롯한 대부분의 프로그래밍 언어는 동시 프로그래밍

(concurrent programming)을 지원하여 이 낭비를 방지한다. 스레드가 어떤 작업을 계속

수행하는 동안 이 스레드에서 다른 스레드를 시작시키고 작업을 수행하게 하는 것이다.

그러나 스레드를 생성하는 것은 쉬울지라도 생성된 스레드는 어떤 이유로든 결국 차단된다.

게다가 다중 스레드로 동시성을 관리하는 것은 쉽지 않다. 스레드가 많을 수록 더 복잡해지기

때문이다.

이에 반해 리액티브 프로그래밍은 본질적으로 함수적이면서 선언적이다. 즉, 순차적으로 

수행되는 작업 단계를 나타낸 것이 아니라 데이터가 흘러가는 파이프라인(pipeline)이나

스트림(stream)을 포함한다. 그리고 이런 리액티브 스트림은 데이터 전체를 사용할 수 있을

때까지 기다리지 않고, 사용 가능한 데이터가 있을 때마다 처리되므로 사실상 입력되는 

데이터는 무한할 수 있다. (이하 책에서 등장하는 비유는 생략한다)

■리액티브 스트림 정의하기

리액티브 스트림은 넷플릭스, 라이트벤드, 피보탈의 엔지니어들에 의해 2013년 말에 

시작되었다. 리액티브 스트림은 차단되지 않는 백 프레셔(backpressure)를 갖는 비동기

스트림 처리의 표준을 제공하는 것이 목적이다.

리액티브 프로그래밍의 비동기 특성은 동시에 여러 작업을 수행하여 더 큰 확장성을 얻게 

해준다. 

한편 backpressure는 데이터를 소비하는 컨슈머가 감당할 수 있을만큼 전달 

데이터를 제한한다. 그래서 지나치게 빠른 데이터 소스로부터의 데이터 전달 폭주를

피할 수 있게 해준다.

자바 스트림 vs 리액티브 스트림

자바 스트림과 리액티브 스트림은 많은 유사성이 있다. 우선, 둘 다 Streams라는 단어가

이름에 포함된다. 또한 데이터로 작업하기 위한 API를 제공한다. 그리고 다수의 똑같은

Operation을 공유한다. 그러나 자바 스트림은 대개 동기화되어 있고, 한정된 데이터로 

작업을 수행한다. 리액티브 스트림은 무한 데이터셋을 비롯해서 어떤 크기의 데이터셋이건

비동기 처리를 지원한다. 그리고 실시간으로 데이터를 처리하며, backpressure를 사용해서

데이터 전달 폭주를 막는다.

리액티브 스트림은 4개의 인터페이스인 Publisher, Subscriber, Subscription, Processor로

요약할 수 있다. Publisher는 하나의 Subscription당 하나의 Subscriber에 발행하는 

데이터를 생성한다. Publisher 인터페이스에는 Subscriber가 Publisher를 구독 신청할 수 

있는 subscribe() 한 개가 선언되어 있다.

package org.reactivestreams;

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

그리고 구독 신청이 완료되면 Subscriber는 Publisher로부터 이벤트를 수신할 수 있다.

이 이벤트들은 Subscriber 인터페이스의 메서드를 통해 전송된다.

package org.reactivestreams;

public interface Subscriber<T> {
    void onSubscribe(Subscription var1);

    void onNext(T var1);

    void onError(Throwable var1);

    void onComplete();
}

Subscriber가 수신할 첫 번째 이벤트는 onSubscribe()의 호출을 통해 이루어진다.

Publisher가 onSubscribe()를 호출할 때 이 메서드의 인자로 Subscription 객체를 

Subscriber에 전달한다. Subscriber는 Subscription 객체를 통해서 구독을 관리할 수 있다.

package org.reactivestreams;

public interface Subscription {
    void request(long var1);

    void cancel();
}

Subscriber는 request()를 호출하여 전송되는 데이터를 요청하거나, 또는 더 이상 데이터를

수신하지 않고 구독을 취소한다는 것을 나타내기 위해 cancel()을 호출할 수 있다.

request()를 호출할 때 Subscriber는 받고자 하는 데이터 항목 수를 나타내는 long 타입의 

값을 인자로 전달한다. 바로 이것이 backpressure이며, Subscriber가 처리할 수 있는 것보다

더 많은 데이터가 Publisher로부터 전송되는 것을 막아준다. 요청된 수의 데이터를 Publisher

가 전송한 후에 Subscriber는 다시 request()를 호출하여 더 많은 요청을 할 수 있다.

---

Subscriber의 데이터 요청이 완료되면 데이터가 스트림을 통해 전달되기 시작한다.

이때 onNext()가 호출되어 Publisher는 Subscriber에게 데이터를 전달하며, 에러가 생기면

onError()가 호출된다. 그리고 Publisher에서 전송할 데이터가 없고 더 이상 데이터를 

생성하지 않는다면 Publisher가 onComplete()를 호출하여 작업이 끝났다고 Subscriber에게

알려준다.

Processor 인터페이스는 다음과 같이 Subscriber 인터페이스와 Publisher 인터페이스를 

결합한 것이다.

package org.reactivestreams;

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

Processor는 데이터를 수신하고 처리하는 Subscriber 역할을 한다. 그리고 데이터의 처리 

결과를 자신의 Subscriber들에게 발행하는 Publisher 역할을 하기도 한다.

즉, Publiser로부터 시작해서 0 또는 그 이상의 Processor를 통해 데이터를 끌어온 다음

최종 결과를 Subscriber에 전달한다.