[Spring] 리액티브 프로그래밍 - 리액티브 스트림즈

 

안녕하세요. 요즘 면접을 보러 가면 비동기 프로그래밍에 대한 질문이 들어오는데 프론트엔드에서의 ajax 개념만 알고 백엔드에서는 몰라서 공부를 해야하나 고민하던 중, 도서관에서 좋은 책을 발견해서 읽고 있습니다. 이에 대해 정리해보려 합니다. 

 

 

이 책입니다. 

https://product.kyobobook.co.kr/detail/S000201399476

스프링으로 시작하는 리액티브 프로그래밍 | 황정식 - 교보문고

 

 

 

리액티브 스트림즈란?

리액티브 스트림즈는 데이터 스트림을 non-blocking 이면서 비동기적인 방식으로 처리하기 위한 리액티브 라이브러리의 표준 사양입니다. 

 

 

 

구성요소

리액티브 스트림즈를 통해 구현해야 하는 API 컴포넌트는 Publisher, Subscriber, Subscription, Processor가 있습니다.

 

Publisher는 데이터를 생성하고 방출하는 역할을 합니다.

Subscribe는 구독한 Publisher로부터 방출된 데이터를 받아서 처리하는 역할을 합니다.

Subscription은 Publisher에 요청할 데이터의 개수를 정하고, 데이터의 구독을 취소하는 역할을 합니다.

Processor는 Publisher와 Subscribe의 기능을 모두 갖고 있습니다.

 

 

Publisher와 Subscribe의 동작 과정

1. Subscribe는 전달받을 데이터를 구독합니다. (subscribe)

2. Publisher는 데이터가 발행될 준비가 되었음을 Subscriber에게 알립니다. (onSubscribe)

3. 준비되었다는 알람을 받은 Subscriber는 전달받기를 원하는 데이터의 개수를 Publisher에게 요청합니다. (Subscription.request)

4. Publisher는 Subscriber로 요청 받은 데이터를 발행합니다. (onNext)

5. Publisher와 Subscriber 간에 데이터 요청 과정을 반복하다가 Publisher가 모든 데이터를 발행하면 마지막으로 데이터 전송이 완료되었음을 Subscriber에게 알립니다. (onComplete) 만약, 처리 과정에서 에러가 발생할 경우 Subscriber에게 알립니다. (onError)

 

 

코드로 보는 리액티브 스트림즈 컴포넌트

Publisher

Kafka의 Pub/Sub 모델과는 달리 리액티브 스트림즈에서의 Publisher와 Subscriber는 개념상 Subscriber가 구독하는 것이 맞으나 코드상으로는 Publisher가 subscribe 메서드의 파라미터인 Subscriber를 등록하는 형태로 이뤄집니다. 

 

Kafka의 경우에는 Publisher와 Subscriber 중간에 메시지 브로커가 있고 브로커에 여러 토픽이 존재하고 Publisher와 Subscriber는 각 브로커 내의 특정 토픽만 바라보게 됩니다. 따라서, Publisher는 특정 토픽으로 메시지 데이터를 전송하고 Subscriber는 특정 토픽을 구독하고 해당 토픽에 전달되는 메시지 데이터를 전달받기만 하면 됩니다. 

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

 

Subscriber

onSubscribe는 구독 시작 시점에 Publisher에게 요청할 데이터의 개수를 지정하거나 구독을 해지하는 것을 말하는 데 이를 메서드 파라미터 Subscription을 통해 이뤄집니다.

onNext는 Publisher가 발행한 데이터를 처리합니다.

onError는 Publisher의 처리 과정에서 에러가 발생했을 때 해당 에러를 처리하는 역할을 합니다.

onComplete는 Publisher가 데이터 통지를 완료햇음을 알릴 때 호출합니다. 

public interface Subscriber<T> {
	public void onSubscribe(Subscription s);
    public void onNext(T t);
    public void onError(Throwable t);
    public void onComplete();
}

 

Subscription

Subscription 인터페이스는 위에서 언급했듯이 Subscriber가 구독한 데이터의 개수를 요청하거나 데이터 요청을 취소하는 역할을 합니다. 

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

 

Processor

Processor는 Publisher와 Subscriber 기능을 모두 갖고 있어 별도로 구현해야할 메서드가 없습니다. 

public interface Processor<T, R> extends Subscriber<T>, Pubisher<R> {

}

 

 

 

관련 용어 정리

Signal

리액티브 스트림즈에서 Publisher와 Subscriber 간에 주고받는 상호작용을 Signal이라고 표현합니다. 

onSubscribe, onNext, onComplete, onError, request, cancel 메서드를 Signal이라고 합니다. 

 

Demand

수요, 요구라는 뜻으로 Subscriber가 Publisher에게 요청하는 데이터를 의미합니다. 

구체적으로는 Publisher가 아직 Subscriber에게 전달하지 않은 subscriber가 요청한 데이터를 말합니다.

 

Emit

Publisher가 Subscriber에게 데이터를 전달하는 것을 데이터를 통지, 발행, 게시, 방출한다는 뜻입니다.

 

Upstream / Downstream

4번 라인의 just 메서드 호출을 통해 반환된 Flux는 5번 라인의 filter 메서드 호출을 통해 반환된 Flux가 자신보다 더 하위에 있어서 Downstream이 됩니다. 

5번 기준에서 4번은 Upstream입니다. 

public class Main {
	public static void main(String[] args) {
    	Flux
        	.just(1, 2, 3, 4, 5, 6(=)
            .filter(n -> n % 2 == 0)
            .map(n -> n * 2)
            .subscribe(System.out::println);
    }
}

 

Sequence

Reactor의 레퍼런스 문서를 참고하다보면 Sequence 용어를 자주 접하게 됩니다. 

Sequence는 Publisher가 Emit 하는 데이터의 흐름을 정의해 놓은 것 자체를 의미합니다. 이 Sequence는 Operator 체인 형태로 정의됩니다.

 

Operator

just, filter, map 과 같은 메서드들을 리액티브 프로그래밍에서 연산자라고 부릅니다. 

 

Source

최초의 가장 먼저 생성된 무언가를 뜻합니다.