[학습] 리액티브 프로그래밍

이 글은 크레이그 윌즈, 『스프링 인 액션』, 심재철 역, 제이펍(2020) 의 내용을 바탕으로 작성하였습니다.

좋은 책이니 스프링의 트렌드를 학습해보고 싶다면 정독을 권합니다.

링크

스프링 인 액션

1. 명령형과 리액티브

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

2. 리액티브 스트림 정의하기

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

리액티브 스트림은 차단되지 않는 백 프레셔를 갖는 비동기 스트림 처리의 표준을 제공하는 것이 목적이다.

자바 스트림 VS 리액티브 스트림 자바 스트림과 리액티브 스트림은 많은 유사성이 있다. 우선, 둘 다 streams라는 단어가 이름에 포함된다. 또한 데이터로 작업하기 위한 API를 제공한다. 둘은 다수의 동일한 오퍼레이션을 공유한다. 그러나, 자바 스트림은 대개 동기화되어 있고 한정된 데이터로 작업을 수행한다. 리액티브 스트림은 무한 데이터셋을 비롯해서 어떤 크기의 데이터셋이건 비동기 처리를 지원한다. 그리고 실시간으로 데이터를 처리하며, 백 프로셔를 사용하여 데이터 전달 폭주를 막는다.

리액티브 스트림은 4개의 인터페이스로 요약할 수 있다.

2.1. Publisher(발행자)

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

Publisher 인터페이스에는 Subscriber가 Publisher를 구독 신청할 수 있는 subscribe() 메서드가 선언되어 있다.

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

2.2. Subscriber(구독자)

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

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

public interface Subscriber<T> {
    void onSubscribe(Subscription sub);
    void onNext(T item);
    void onError(Throwable ex);
    void onComplete();
}

2.3. Subscription(구독)

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

Publisher가 onSubscribe()를 호출할 때 이 메서드의 인자로 Subscription 객체를 Subscriber에 전달한다.

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

public interface Subscription{ 
    void request(long n); 
    void cancle(); 
}

Subscriber는 request()를 호출하여 전송하는 데이터를 요청하거나, 또는 더 이상 데이터를 수신하지 않고 구독을 취소한다는 것을 나타내기 위해 cancle()을 호출할 수 있다.

request()를 호출할 때 Subscriber는 받고자 하는 데이터 항목 수 n(long)을 인자로 전다랗ㄴ다.

이것이 바로 백 프레셔이며, Subscriber가 처리할 수 있는 것보다 더 많은 데이터를 Publisher가 전송하는 것을 막아준다.

요청한 데이터를 전송받은 뒤, Subscriber는 다시 request()를 호출하여 더 많은 요청을 할 수 있다.

---

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

이때 onNext() 메서드가 호출되어 데이터가 전달되고, 에러 발생 시 onError()가 호출된다.

또한 데이터가 더 이상 없을 경우 onComplete()를 호출하여 작업이 끝난 것을 Subscriber에 알린다.

2.4. Processor(프로세서)

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

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

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

그 다음에 역할을 바꾸어, Publisher 역할로 처리 결과를 자신의 Subscriber들에게 발행한다.

---

우리는 이를 통해, Publisher에서 시작하여 0개 이상의 Processor를 거쳐,

Subscriber에 최종 결과를 전달하는 파이프라인을 만들 수 있다.

그러나 리액티브 스트림 인터페이스는 스트림을 구성하는 기능이 없다.

이에 따라 프로젝트 리액터에서는 리액티브 스트림을 구성하는 API를 제공하여 리액티브 스트림 인터페이스를 구현하였다.

이는 스프링5의 리액티브 프로그래밍 모델의 기반이다.