Reactive Stream 을 사용하기 위해서는 다음과 같은 것을 구현해줘야 한다.

• Publisher
• Subscriber
• Subscription

한가지 문제가 있다. 이것을 구현하기가 쉽지가 않다. 특히 Publisher 의 경우에는 Reactive Stream 의 기본 아이디어만 가지고 간단하게 구현할 수가 없다. 어떤 것을 Publisher 할지도 영향을 주는것도 문제지만 Reactive 의 Spec 을 마춰짜야 하는데 이게 쉽지가 안다.

Publisher 를 간단하게 구현 방법이 없을까? SubmissionPublisher 를 이용하면 간단하게 사용해 볼 수 있다.

이 문서는 SubmissionPublisher 를 이용해 어떻게 Reactive Stream 을 구현하는지를 설명한다.

SubmissionPublisher Class

이 클래스를 간단히 살펴보자.

이 클래스는 Publisher 인터페이스를 구현한 구현체 클래스 이다. 하지만 아주 간단하지만은 않다.

SubmissionPublisher 클래스가 가지는 메소드가 꽤 있다. 자세히 보면 get 혹은 is 로 시작하는 메소드가 많음을 알 수 있다. 이 클래스들은 Publisher 의 상태 정보를 체크하기 위한 것들이다. 반대로 set 으로 시작하는, 그러니까 뭔가를 지정하는 메소드는 많지가 않다.

이 클래스에 설명은 다음과 같이 시작한다.

A Flow.Publisher that asynchronously issues submitted (non-null) items to current subscribers until it is closed. Each current subscriber receives newly submitted items in the same order unless drops or exceptions are encountered. Using a SubmissionPublisher allows item generators to act as compliant reactive-streams Publishers relying on drop handling and/or blocking for flow control.

Flow.Publisher 는 Null 이 아닌 제출된 아이템들을(item) 현재 구독자에게 닫힐때까지 비동기적으로 발행한다. 각각의 현재 구독자들은 삭제 또는 예외가 발생하지 않는한 새롭게 제출된 아이템들을 같은 순서로 받는다. SubmissionPublisher 를 사용하는 것은 아이템 생성기를 흐름제어를 위해 삭제 핸들링과 블럭킹을 필요로하는 reactive-streams Publisher 와 호환되는 것처럼 행동하게 해준다.

https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/SubmissionPublisher.html

어찌되었든 간에 이것을 이용하면 Reactive Streams Publisher 처럼 만들어 준다 것에 주목해야 한다.

SubmissionPublisher 사용하기

사용방법은 아주 간단하다.

SubmissionPublisher 의 객체를 하나 생성한다. 그리고 Reactive Stream 이 Publisher 에 요구하는 subscribe 메소드를 호출해 준다.

하지만 subscribe() 메소드는 인자값으로 Subscriber 클래스를 요구한다. Subscriber 클래스는 Interface 클래스여서 구현체 클래서를 넣어야 한다. 다양한 방법이 있지만 여기서는 다음과 같이 작성해 본다.

Reactive Stream 이 요구하는 Subscriber 에 요구 메소드들이(onNext, onError, onComplete) 모두 나온다. 다음과 같이 모두 구현해 준다.

위와같이 구현을 해주면 된다. 여기서 중요한 것이 존재하는데, Subscription 객체 변수를 반드시 써야 한다는 것이다. onSubscribe, onNext 메소드에서 공통으로 보이는 subscription.request(1) 이 바로 그것인데 onNext 메소드에 이를 사용하기 위해서 Subscriber 구현체에서 멤버변수를 만들어두고 onSubscribe 시에 주입된 객체변수를 할당해 준다.

이제 다음과 같이 publisher 의 submit 메소드를 사용해 아이템을 넣어준다.

submit 메소드는 아이템을 제출해 비동기적으로 Subscriber 에게 생성해준다. SubmissionPublisher 의 비동기는 Thread 를 이용한다. 그래서 모든 Thread 가 완료되기를 Main Thread 를 몇초간 중지줘야해서 Thread.sleep(100) 코드가 필요하다.

모든 비동기 제출이 완료되면 제출자를 닫아준다. publisher.close() 다.

이렇게 완료된 전체 코드는 다음과 같다.

Reactive Stream 을 어떻게하면 간단하게 사용해볼까. Reactive Stream 이 요구하는 최소사항을 어떻게하면 직접 사용해볼 수 있을까? 그렇다면 SubmissionPublisher 클래스를 활용하라.

자바(Java) 에서 Reactive Stream 구현체를 살펴본다.

이전 글에서 Reative Stream 의 기본 아이디어를 설명했었다.

• Flow control
• Publisher – Subscriber pattern

또한 이 아이디어를 위한 포맷은 Publisher – Subscriber 이다. Reactive Stream 에서 이 모델은 1) Subscriber 가 Publisher 에게 가입을 요청하면 2) Publisher 는 Subscriber 와 통신을 위한 채널인 Subscription 을 생성하고 이를 통보한다. 이는 다음 그림과 같이 묘사할 수 있다.

여기서 주목해야할 것이 Subscriber 는 Subcription 을 통해서 데이터를 요청하고 받게 된다는데 있다.

java.util.concurrent.Flow

자바에서 구현은 위 클래스에 구현 되어 있다. 그리고 이 클래스 첫줄에 다음과 같이 설명이 시작된다.

Interrelated interfaces and static methods for establishing flow-controlled components in which Publishers produce items consumed by one or more Subscribers, each managed by a Subscription

흐름 제어 컴포넌트를 설정하기 위한 (밀접하게) 상호연관된 인터페이스(Interface) 와 스태틱 메소드(Static method)

java.util.concurrent.Flow

주석에 내용에 등장하는 컴포넌트 Publisher, Subscriber, Subscription 이다. Flow.class 는 다음과 같이 구성되어 있다.

Publisher

FunctionalInterface 인게 눈에 뛴다. 이것은 오직 한개의 메소드만을 가진다는 것과 Lambda 식을 지원하게 된다. subscribe 메소드는 인자로 Subscriber 를 받는다.

중간에 주석에 주목할만한 내용이 나온다.

Subscribers may enable receiving items by invoking the request method of this Subscription, and may unsubscribe by invoking its cancel method.

Subscriber 는 이 Subcription 의 request 메소드를 호출함으로써 아이템들을 수신할 수 있으며 cancel 메소드를 호출함으로써 가입이 해제된다.

Subcriber 는 Subscription 의 request 메소드를 호출함으로써 데이터(아이템)을 수신한다는 것이다.

Subscriber

4개의 메소드를 가지고 있다. 이 메소드들이 다루는 것이 무엇인지를 아는것이 제일 중요하다. 각 메소드의 주석에는 Subscription 이 항상 나온다.

Method invoked with a Subscription’s next item. If this method throws an exception, resulting behavior is not guaranteed, but may cause the Subscription to be cancelled

Subscription’s 의 다음 아이템을 호출하는 메소드. 만약 이 메소드가 예외를 발생시키면, 최종 결과를 보장하지 않지만, Subcription 은 취소될 수 있다.

onNext

Subscriber 가 다음 아이템을 호출하는데 Publisher 를 대상으로 하지 않고 Subscription 을 대상으로 한다는 것을 꼭 기억해야 한다.

Subscription

Publisher 와 Subscriber 를 연결하는 메시지 제어를 하는게 목적이다 .request 메소드는 Subcription 에 대해 현재 처리되지 않은 요청을 위한 아이템의 개수를 추가하게 된다. 몇개의 아이템을 요청할 것인지를 요청하는 것이다.

SubmissionPublisher

SubmissionPublisher 는 non-null 아이템을 생성해주는 Publisher 이다. 생성을 해준다고 해서 자동으로 생성해주는 것이 아니라 아이템을 제출해야 한다. 이 제출된 아이템은 순차적으로 Publish 해준다.

submit

이 메소드는 비동기적으로 Subscriber의 onNext 메소드를 호출함으로써 인수로 받은 아이템을 Subscriber 에 아이템을 발행한다.

Reactive Stream 에는 다양한 프레임워크, 라이브러리들이 존재한다. 이것들은 각자의 고유한 이름과 특징을 가지고 있다. 어떤 것들이 있는지 간단하게 정리해 본다.

Reactive Streams

저수준의 규약으로 주로 자바 인터페이스(Interface) 로 구현 되어 있다. 명시적인 back-pressure 으로 Publisher 와 Subscriber 의 기본 빌딩 블록을 표현한다. Java 9 에서 java.util.concurrent.Flow 로 구현되어 있다.

RxJava

이것은 Reactive Extension 이다. Neflix 에서 개발해 오픈소스로 전화하면서 세상에 알려졌다. ReactiveX 라고도 불린다. 브릿지 Reactive Stream 이라고 말하기도 하는데, Reactive Stream 을 위한 타입 전환을 지원한다.

Reactor

자바 프레임워크 이다. Spring 배포로 유명한 피보탈(Pivotal) 에서 제작해 오픈소스로 배포하고 있다.

Spring Framework

HTTP 서버/클라이언트을 툴을 비롯해 리액트 기능을 가지고 있다. Spring 의 특징은 어노테이션을 이용한 기능 주입등을 사용할 수 있다. 비동기 HTTP 서버에서 구동가능하다. (Netty, Undertow, Tomcat 8 이상)

Akka Stream

Actor Model 을 기반으로 애플리케이션 개발을 위한 Java, Scala 를 위한 툴킷이다.

이것을 정리하는 이유가 있다. Reactive Stream 은 스펙이기 때문이다. 이 스펙을 만족하면 Reactive Stream 이라고 할 수 있다. 물론 자바의 저수준 인터페이스 이름도 Reactive Stream 이라 헷깔릴 수 있지만 Reactive Stream 스펙을 만족하는 다양하고 라이브러리, 프레임워크, 툴킷들이 있다.