리액티브 스트림즈(Reactive Streams)란?

이 글은 인프런의 Kevin의 알기 쉬운 RXJava 1부를 참고합니다.

데이터를 생성해서 내보는 쪽 : 생산자
생산자가 데이터를 내보는 것을 통지
데이터를 소비하는 것을 소비자

Reactive Streams란?

• 리액티브 프로그래밍 라이브러리의 표준사양입니다.
  https://github.com/reactive-streams/reactive-streams-jvm/

• 리액티브 프로그래밍에 대한 인터페이스만 제공합니다.

• RxJava는 이 Reactive Streams의 인터페이스들을 구현한 구현체입니다.

• Reactive Streams는 Publisher, Subscriber, Subscription, Processor라는 4개의 인터페이스를 제공합니다.
  - Publisher : 데이터를 생성하고 통지합니다.
  - Subscriber : 통지된 데이터를 전달받아서 처리합니다.
  - Subscription : 전달 받을 데이터의 개수를 요청하고 구독을 해지합니다.
  - Processor : Publisher와 Subscriber의 기능이 모두 있습니다.
  

Publisher와 Subscriber간의 프로세스 흐름

Cold Publisher & Hot Publisher

Cold Publisher

• 생산자는 소비자가 구독할때마다 데이터를 처음부터 새로 통지합니다.
• 데이터를 통지하는 새로운 타임라인이 생성됩니다.
• 소비자는 구독시점과 상관없이 통지된 데이터를 처음부터 전달 받을 수 있습니다.

Hot Publisher

• 생산자는 소비자 수와 상관없이 데이터를 한번만 통지합니다.
• 즉, 데이터를 통지하는 타임라인은 하나입니다.
• 소비자는 발행된 데이터를 처음부터 전달 받는게 아니라 구독한 시점에 통지된 데이터들만 전달받을 수 있습니다.

코드로 확인해보기

Cold Publisher

public class ColdPublisherExample {
    public static void main(String[] args){
        Flowable<Integer> flowable = Flowable.just(1, 3, 5, 7);

        flowable.subscribe(data -> System.out.print("구독자1: " + data + " "));
        System.out.println();
        flowable.subscribe(data -> System.out.print("구독자2: " + data + " "));
    }
}

결과는 아래와 같습니다.

첫번째 구독자도 1,3,5,7의 데이터를 전달받고
두번째 구독자도 1,3,5,7의 데이터를 전달받았습니다.

Hot Publisher

public class HotPublisherExample {
    public static void main(String[] args){
        PublishProcessor<Integer> processor = PublishProcessor.create();
        processor.subscribe(data -> System.out.println("구독자1: " + data));
        processor.onNext(1);
        processor.onNext(3);

        processor.subscribe(data -> System.out.println("구독자2: " + data));
        processor.onNext(5);
        processor.onNext(7);

        processor.onComplete();
    }
}

결과는 아래와 같습니다.

구독자 1은 1,3,5,7의 데이터를 전달받았고
구독자 2는 5,7의 데이터를 전달받았습니다.

결론

Reactive Streams에 대해 알아보았고 핵심요소인 Publisher 중에서 Hot Pubisher와 Cold Publisher에 대해 알아보았습니다.