[📕 Reactive] 2. Reactive Streams

 

1. 리액티브 스트림즈란 ? 

데이터 스트림을 Non-Blocking이면서 비동기적인 방식으로 처리하기 위한 리액티브 라이브러리의 표준 사양

 

✳️ publisher와 subscriber 간에 데이터 전달 동작 과정 

1. subscriber는 전달받을 데이터 구독 (subscribe)
2. publisher는 데이터 통지할 준비되었음을 subscriber에 알림 (onSubscribe)
3. publisher가 데이터를 통지할 준비가 되었다는 알림을 받은 subscriber는 전달받기를 원하는 데이터의 개수를 publisher에게 요청 (Subscription.request)
4. publisher는 subscriber로부터 요청받은 만큼의 데이터를 통지함 (onNext)
5. publisher와 subscriber 간에 데이터 통지, 데이터 수신, 데이터 요청의 과정을 반복하다가 publisher가 모든 데이터를 통지하게 되면 마지막으로 데이터 전송이 완료되었음을 subscriber에게 알림 (onCompelete)
6. 만약 publisher가 데이터를 처리하는 과정에서 에러가 발생하면 에러가 발생했음을 subscriber에게 알림 (onError)

 

 

 

 

2. 리액티브 스트림즈 컴포넌트

✳️ publisher

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

⛔ subscriber가 구독하는 것인데, 왜 publisher 인터페이스에 subscribe 메서드가 정의되어있음?

👉 개념상으로는 그게 맞지만 publisher가 subscriber를 등록하는 형태로 구독이 이루어짐. 

 

 

✳️ subscriber

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

• onSubscribe : 구독 시작 시점에 어떤 처리(publisher에게 요청할 데이터 개수 지정하거나 구독 해지)를 하는 역할
• onNext : publisher가 통지한 데이터 처리
• onError : 데이터 처리 과정에서 에러 발생했을 때 해당 에러 처리 
• onComplete : 데이터 통지가 정상적으로 완료됐음을 알림 

 

✳️ subscription

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

• subscriber가 구독한 데이터의 개수 요청하거나 데이터 요청 취소 (구독 해지)

 

✳️ processor

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

}

• subscriber 인터페이스와 publisher 인터페이스 상속
• processor가 publisher와 subscriber 기능을 모두 가지고 있음 

 

 

 

 

3. 리액티브 스트림즈 구현 규칙

 

✳️ publisher 구현 주요 기본 규칙 

1. subscriber가 요청한 것보다 더 많은 개수의 데이터를 publisher가 보낼 수 없음
2. publisher는 요청된 것보다 적은 수의 onNext signal을 보내고, onComplete 또는 onError를 호출해 구독 종료 가능
3. publisher가 전송한 onError 또는 onComplete signal은 구독 취소와 동일한 기능 수행

 

✳️ subscriber 구현 주요 기본 규칙

1. subscriber가 처리할 수 있는 적절한 상한선만큼의 데이터 개수 요청 (subscription.requets(n))
2. subscriber.onComplete() 또는 subscriber.onError(Throwable t)는 subscription 또는 publisher의 메서드를 호출해서는 안됨 👉 publisher/subscription과 subscriber 간의 순환 및 경쟁 조건 방지하기 위함 

 

✳️ subscription 구현 주요 기본 규칙

1. subscriber가 onNext와 onSubscribe 내에서 동기적으로 Subscription.request를 호출하도록 허용해야 함 
2. subscription.cancel()을 호출했을 때, publisher가 signal 전송을 중지할뿐만 아니라 구독 취소를 요청한 구독자에 대한 참조까지 삭제함
3. 구독자는 한 번 요청할 때 무한한 개수의 데이터 요청 가능하고, 그 요청을 끝없이 호출 가능