모던 자바 인 액션 -15-

리액티브 프로그래밍 패러다임의 중요성이 증가하는 이유

• 빅데이터 : 매일 증가하는 데이터의 양
• 다양한 환경 : 모바일 디바이스에서 클라우드 클러스터에 이르기까지 다양한 환경에 애플리케이션
• 사용 패턴 : 사용자는 1년 내내 항상 서비스를 이용할 수 있으며 짧은 응답 시간을 기대한다

리액티브 프로그래밍에서는 다양한 시스템과 소스에서 들어오는 데이터 항목 스트림을 비동기적으로 처리하고 합쳐서 이런 문제를 해결한다

 

리액티브 매니패스토 

리액티브 애플리케이션과 시스템 개발의 핵심 원칙

• 반응성 : 리액티브 시스템은 빠를 뿐 아니라 일정하고 예상할 수 있는 반응 시간을 제공한다
• 회복성 : 장애가 발생하더라도 시스템은 반응해야 한다. 리액티브 매니페스토는 회복성을 달성할 수 있는 다양한 기법을 제공한다
• 탄력성 : 다양한 작업 부하 속에서 반응성을 유지한다
• 메시지 주도 : 회복성과 탄력성을 지원하려면 시스템을 구성하는 컴포넌트의 경계가 명확해야 정의해야한다. 비동기 메시지에 의존해 컴포넌트 끼리 통신이 이루어져야 한다.

자바9에서는 리액티브 프로그래밍을 제공하는 클래스 java.util.concurrent.Flow를 추가했다

이 클래스는 정적 컴포넌트 하나를 포함하고 있으며 인스턴스화 할 수 없다.

 

Flow 클래스는 중첩된 인터페이스 네 개를 포함한다

• Publisher
• Subscriber
• Subscription
• Processor

Publisher가 항목을 발행하면 Subscriber가 한 개씩 또는 한 번에 여러 항목을 소비하는데 Subscription이 이 과정을 관리할 수 있도록 Flow 클래스는 관련된 인터페이스와 정적 메서드를 제공한다.

Subscriber 요구사항에 따라 역압력 기법에 의해 이벤트 제공 속도가 제한된다

Publisher는 자바의 함수형 인터페이스로, Subscriber는 Publisher가 발행한 이벤트의 리스너로 자신을 등록할 수 있다.

Subscription은 Publisher와 Subscriber 사이의 제어 흐름, 역압력을 관리한다

 

Publisher 인터페이스

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

Subscriber 인터페이스

public interface Subscriber<T>{
  // 처음 호출시
	void onSubscribe(Subscription s);
  // 여러번 onNext
  void onNext(T t);
  //error 발생시
  void onError(Throwable t);
  // 구독 종료
  void onComplete();
}

 

이들 이벤트는 다음 프로토콜에서 정의한 순서로 지정된 메서드 호출을 통해 발행되어야 한다

https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.3/README.md#3.5

onSubscribe onNext* (onError | onComplete)?

위 표기는 onSubscribe 메서드가 항상 처음 호출되고 이어서 onNext가 여러번 호출될 수 있다. 

onComplete 콜백을 통해 더이상의 데이터가 없고 종료됨을 알릴 수 있으며 또는 Publisher에 장애가 발생했을 때는 onError 를 호출할 수 있다.

 

Subscriber가 Publisher에 자신을 등록할 때 Publisher는 처음으로 onSubscribe 메서드를 호출해 Subscription 객체를 전달한다. 

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

Subscription 인터페이스는 request 메서드로 Publisher에게 주어진 개수의 이벤트를 처리할 준비가 되었음을 알릴 수 있다. cancel 메서드로 Subscription을 취소, 즉 Publisher에게 더 이상 이벤트를 받지 않음을 통지한다

 

자바 플로 규칙 

• Publisher는 반드시 Subscription의 request 메서드에 정의된 개수 이하의 요소만 Subscriber에 전달해야한다. Publisher는 지정된 개수 보다 적은 수의 요소를 onNext로 전달할 수 있으며 동작이 성공적으로 끝났으면 onComplete를 호출하고 문제가 발생하면 onError를 호출해 Subsciption을 종료할 수 있다
• Subscriber는 요소를 받아 처리할 수 있음을 Publisher에 알려야한다. 이런 방식으로 Subscriber는 Publisher에 역압력을 행사할 수 있고 Subscriber가 관리할 수 없이 너무 많은 요소를 받는 일을 피할 수 있다. 더욱이 onComplete나 onError 신호를 처리하는 상황에서 Subscriber는 Publisher나 Subscription의 어떤 메소드도 호출할 수 없으며 Subscription이 취소되었다고 가정해야한다. 마지막으로 Subscriber는 Subscription, request() 메서드 호출이 없이도 언제든 종료 시그널을 받을 준비가 되어있어야 하며 Subscription.cancel()이 호출된 이후에라도 한 개 이상의 onNext를 준비가 되어있어야한다.
• Publisher와 Subscriber는 정확하게 Subscription을 공유해야하며 각각이 고유한 역할을 수행해야한다. 그러려면 onSubscribe와 onNext 메서드에서 Subscriber는 request메서드를 동기적으로 호출할 수 있어야 한다. 표준에서는 Subscription.cancel()메소드는 몇번을 호출해도 한번 호출한 것과 같은 효과를 가져야하며, 여러번 이 메소드를 호출해도 다른 추가 호출에 별 영향이 없도록 스레드에 안전해야 한다고 명시한다. 

폴로 API를 사용하는 리액티브 애플리케이셔 라이프 사이클

Flow 클래스의 Processor 인터페이스는 리액티브 스트림에서 처리하는 이벤트의 변환단계를 나타낸다.Processor가 에러를 수신하면 이로부터 회복하거나(그리고 Subscription은 취소로 간주) 즉시onError 신호로 모든 Subscriber에 에러를 전파 할 수 이다.

 

자바 9 폴로 API

자바 라이브러리는 보통 인터페이스와 구현을 제공하지만 폴로 API는 구현을 제공하지 않는다. 이는 API 당시 Akka, RxJava 등 다양한 리액티브 스트림의 자바 코드 라이브러리가 존재 했기 때문이다. 폴로 API는 다양한 라이브러리의 표준화 작업을 해준 것이다. 

 

리액티브 라이브러리 RxJava

RxJava에서는 Flowable 클래스(역압력 기능 있음) 와 Observable 클래스(역압력 기능 없음)를 제공한다. 

 

간단한 Obseravble 만들기( just는 java의 of와 비슷하다)

Observable<String> strings = observable.just("first", "second");

Observable의 구독자는 onNext("first"), onNext("second"), onComplete()의 순서로 메시지를 받는다

 

Observable이 플로 API의 Publisher 역할을 하며 Observer는 Subscriber 인터페이스 역할을 한다.

다른 것은 Subscription 대신 Disposable 인수를 가진다.

 

마블 다이어그램 

위의 실선 : 변환 이전의 Observable의 시간선(왼쪽에서 오른쪽으로 흐름) 

아래의 실선 : 변환 결과 Observable의 시간선(왼쪽에서 오른쪽으로 흐름)

X : 어떤 이유로 Observable이 비정상 종료되어 에러 발생하면 수직선이 X로 바뀐다

점선과 상자 : Observable에 적용되는 변환을 가리킨다. 상자 안의 텍스트는 변환의 종류를 가르킨다

도형 : Observable이 방출하는 항복이다

세로선 : Observable이 성공적으로 완료되었음을 가르킨다

 

*message 기반 vs event 기반 차이

 

메시지 기반은 producer가 응답이나 앞으로 일어날 명령등에 대해 반드시 알고 있다면 messaging 기반, 즉 누구에게 보낼지, 어떤행동을 보낼지 알고있다면 message 기반

 

반면 이벤트 기반은 응답이나, 앞으로 일어날 명령등에 상관없이 producer가 event를 발행, 그냥 어떠한 action이 발생하면 listening 하고 있는 컴포넌트 모두에게 data를 발행하는 것