Reactive Streams is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure.
Reactive Streams란 non-blocking과 back pressure를 이용한 asynchronous 스트림 처리의 표준이다.
중요한 키워드가 여러개 등장 했는데, 먼저 back pressure에 대해서 알아보자.
Back Pressure
Back Pressure는 Reactive Streams에서 가장 중요한 요소라고 할 수 있다. Back Pressure가 등장하게 된 배경을 이해하기 위해서 먼저 옵저버 패턴을 이해하고 옵저버 패턴이 갖고 있는 문제점을 인식할 수 있어야한다.
Observable & Observer
옵저버 패턴(observer pattern) 은 객체의 상태 변화를 관찰하는 관찰자들, 즉 옵저버들의 목록을 객체에 등록하여 상태 변화가 있을 때마다 메서드 등을 통해 객체가 직접 목록의 각 옵저버에게 통지하도록 하는 디자인 패턴이다. 주로 분산 이벤트 핸들링 시스템을 구현하는 데 사용된다. 발행/구독 모델로 알려져 있기도 하다.
@Override publicvoidrun(){ for (int i = 1; i <= 10; i++) { setChanged(); notifyObservers(i); // push } } }
publicstaticvoidmain(String[] args){ Observer ob = new Observer() { @Override publicvoidupdate(Observable o, Object arg){ log.info("{}", arg); } };
IntObservable io = new IntObservable(); io.addObserver(ob);
ExecutorService es = Executors.newSingleThreadExecutor(); es.execute(io);
log.info("EXIT"); es.shutdown(); } }
- 실행결과 01:47:30.715 [main] INFO com.jongmin.reactive.practice.Ob - EXIT 01:47:30.715 [pool-1-thread-1] INFO com.jongmin.reactive.practice.Ob - 1 01:47:30.719 [pool-1-thread-1] INFO com.jongmin.reactive.practice.Ob - 2 01:47:30.719 [pool-1-thread-1] INFO com.jongmin.reactive.practice.Ob - 3 01:47:30.719 [pool-1-thread-1] INFO com.jongmin.reactive.practice.Ob - 4 01:47:30.719 [pool-1-thread-1] INFO com.jongmin.reactive.practice.Ob - 5 01:47:30.719 [pool-1-thread-1] INFO com.jongmin.reactive.practice.Ob - 6 01:47:30.719 [pool-1-thread-1] INFO com.jongmin.reactive.practice.Ob - 7 01:47:30.719 [pool-1-thread-1] INFO com.jongmin.reactive.practice.Ob - 8 01:47:30.719 [pool-1-thread-1] INFO com.jongmin.reactive.practice.Ob - 9 01:47:30.719 [pool-1-thread-1] INFO com.jongmin.reactive.practice.Ob - 10
문제점
옵저버 패턴에서는 Publisher(Observable)이 Subscriber(Observer)에게 데이터(이벤트)를 Push(notifyObservers)하는 방식으로 전달한다. 이때, Publisher는 Subscriber의 상태에 상관없이 데이터를 전달하는데만 집중한다.
만약, Subscriber는 1초에 10개의 데이터를 처리할 수 있는데 Publisher가 1초에 20개의 데이터를 전달(Push)한다면 어떤 문제가 발생할까? 다음과 같은 문제가 발생할 수 있다.
Subscriber에 별도의 queue(버퍼)를 두고 처리하지 않고 대기중인 데이터를 저장할 수 있다.
하지만, queue의 사용 가능한 공간도 전부 금방 소모될 것이다.
queue의 크기를 넘어가게 되면 데이터는 소실될 것이다.
queue의 크기를 너무 크게 생성하면 OOM(Out Of Memory) 문제가 발생할 수 있다.
해결 방법
Observable과 Observer의 문제를 어떻게 해결할 수 있을까? Publisher가 Subscriber에게 데이터를 Push 하던 기존의 방식을 Subscriber가 Publisher에게 자신이 처리할 수 있는 만큼의 데이터를 Request하는 방식으로 해결할 수 있다. 필요한(처리할 수 있는) 만큼만 요청해서 Pull하는 것이다. 데이터 요청의 크기가 Subscriber에 의해서 결정되는 것이다. 이를 dynamic pull 방식이라 부르며, Back Pressure의 기본 원리이다.
Reactive Streams API
Reactive Streams는 표준화된 API이다. 2013년 netflix, pivotal, lightbend의 엔지니어들에 의해서 처음 시작되어, 2015 4월에 JVM에 대한 1.0.0 스펙이 릴리즈 되었다. Java 9부터는 reactive streams이 java.util.concurrent의 패키지 아래 Flow라는 형태로 JDK에 포함되었다. 기존에 reactive streams가 가진 API와 스펙, pull방식을 사용하는 원칙을 그대로 수용하였다.
아래는 Reactive Streams API이다.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
publicinterfacePublisher<T> { publicvoidsubscribe(Subscriber<? super T> s); }
Publisher: Subscriber를 받아들이는 subscribe 메서드 하나만 갖는다.
Subscriber: 데이터를 받아 처리할 수 있는 onNext, 에러를 처리하는 onError, 모든 데이터를 받아 완료되었을 때는 onComplete, 그리고 Publisher로부터 Subscription을 전달 받는 onSubscribe 메서드로 이루어진다.
Subscription: n개의 데이터를 요청하는 request와 구독을 취소하는 cancel을 갖는다.
전체적인 흐름은 다음과 같다.
Subscriber가 Publisher에게 구독을 요청한다.
Publisher는 Subscriber의 onSubscribe 메서드를 통해 Subscription을 전달한다.
Subscriber는 Publisher에게 직접 데이터를 요청하지 않고 Subscription을 통해 요청한다.
Publisher는 Subscription을 통해 onNext에 데이터를 전달하고 완료되면 onComplete, 에러가 발생하면 onError에 전달한다.
Example
마지막으로 Reactive Streams API를 간단하게 구현해 테스트 해보자.
Reactive Streams API의 Interface는 간단해 보이지만 이를 구현한 구현체는 Reactive Streams Specification을 만족해야만 한다. 구현체가 Specification을 만족하는지는 Reactive Streams TCK(Technology Compatibility Kit)라는 도구를 이용해 검증할 수 있다.