RxJava란? - 기본 구조

RxJava는 Reactive java에서 이름을 따왔다고 한다.

Reactive programming 패러다임을 자바에서 구현한 프로그래밍 라이브러리이다.

프로그래밍 패러다임에는 여러가지가 있는데 OOP(객체지향), Function(함수), Imperative(명령형) 등이 있다.

여러가지 언어 중에서 자바는 OOP(Object Oriented Programming)라고 해서 객체지향형 프로그래밍의 대표 언어이고 Functional Programming 을 대표하는건 파이썬, node.js 하스켈, 스칼라, 클로저등 이라고 할 수 있다. 자바로 Reactive Programming을 해야할 일이 생겨서 이 라이브러리가 등장 하였다. 그리고 자바가 버젼이 올라가면서 여러가지 요즘 트렌트에 맞게 기술들이 추가 되어서 OOP기반이었던 자바가 Functional, Reactive 등의 프로그래밍 방법론으로도 개발이 가능하게 되었다.

리액티브란 외부에서 자극이 오고 그에 대해 반응 한다는 뜻이다.

데이터의 강을 만드는 옵저버블(Observable)과 강에서 데이터를 하나씩 건지는 서브스크라이버(Subscriber)가 리액티브 프로그래밍의 가장 핵심적인 요소이다.

Hello, World!

이제 구현된 예제로 이 프레임워크가 동작하는 것을 보겠습니다. 먼저 기본적인 Observable을 생성해보겠습니다 :

Observable<String> myObservable = Observable.create(
  new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> sub) {
      sub.onNext(“Hello, world!”);
      sub.onCompleted();
    }
  }
);

이 Observable은 데이터를 제공하는 생산자로 세가지 유형의 행동을 합니다.

1. onNext - 새로운 데이터를 전달한다.
2. onCompleted - 스트림의 종료
3. onError - 에러 신호를 전달한다.

 Observable은 "Hello, world!”를 전달하고(onNext) 다음으로 끝났다는 신호(onCompleted)를 전달합니다. Observable의 세가지 행동들이 만드는 스트림을 도식 해보면 다음과 같습니다.

 이제 데이터를 소비하는 Subscriber을 생성해보겠습니다:

Subscriber<String> mySubscriber = new Subscriber<String>() {
  @Override
    public void onNext(String s) { System.out.println(s); }
  @Override
    public void onCompleted() { }
  @Override
    public void onError(Throwable e) { }
};

이것은 Observable이 만드는 스트림에 응대하여 처리할 수 있게 대칭적으로 인터페이스가 구성되어 있습니다.

이제 우리는 myObservable과 mySubscriber을 가졌으며 subscribe()를 사용하여 서로를 연결시킬 수 있습니다 :

myObservable.subscribe(mySubscriber);
// Outputs “Hello, world!”

구독(subscription)을 만들면 myObservable은 subscriber의 onNext()와 onComplete() 메소드를 호출합니다. 그 결과 mySubscriber는 "Hello, world!”를 출력한 뒤 종료됩니다.

Simpler Code

단지 "Hello, World"를 출력하면서 너무 많은 코드가 작성 되어있습니다. 이것은 이해를 위하여 작성 되었으며 RxJava는 코드 작성을 쉽게 만들어 주는 지름길을 많이 제공합니다..

일단 Observable을 간소화하자. RxJava는 일반적인 작업을 위한 Observable 생성 메소드들을 여러개 내장하고 있습니다.

아래 코드 경우에는 Observable.just()가 아이템을 하나 발행한 뒤 종료합니다.

Observable<String> myObservable =
  Observable.just(“Hello, world!”);

엄밀하게 말하면, Observable.just()는 우리의 초기 코드와 정확히 동일하지는 않습니다. 그 이유는 다음 글에서 작성하도록 하겠습니다.

다음으로는 불필요하게 장확한 Subscriber를 다뤄보겠습니다.  위에 코드에서는 onNext()에서 무엇을 수행할지를 정의하였지 onCompleted()와 onError() 에 둘 다 코드를 작성하지 않았습니다.

RxJava에서는 편의를 위해 사용하지 않는 구성이 누락된 인터페이스를 제공합니다.

Action1<String> onNextAction = new Action1<String>() {
  @Override
  public void call(String s) {
    System.out.println(s);
  }
};

액션들은 Subcriber의 각 파트를 정의 할 수 있습니다. Observable.subscribe()는 onNExt(), onError() 그리고 onComplete()를 대신할 하나, 둘 또는 세 개의 액션 파라미터를 다룰 수 있습니다. 

이전에 선언한 Subscriber는 다음과 같이 생겼습니다.

myObservable.subscribe(onNextAction, onErrorAction, onCompleteAction);

하지만  onError()와 onComplete()를 무시할 것이므로 첫 번쨰 파라미터만 필요합니다.

myObservable.subscribe(onNextAction);
// Outputs “Hello, world!”

이제 메소드 호출을 서로 연쇄(chaining) 시켜 변수들을 제거합니다.

Observable.just(“Hello, world!”)
  .subscribe(new Action1<String>() {
    @Override
    public void call(String s) {
      System.out.println(s);
    }
  });

마지막으로 Action1코드를 제거하기 위해 JAVA8의 람다를 이용합니다.

Observable.just(“Hello, world!”)
 .subscribe(s -> System.out.println(s));