[RxJava] RxJava란?

RxJava가 등장하게 된 배경

비동기 결과를 또 비동기로 써야할 때 문제가 발생.

 

예시)

1. 로그인 요청 (네트워크)
2. 로그인 성공하면 → 채팅방 목록 요청
3. 채팅방 선택하면 → 메시지 불러오기
4. 메시지 오면 → 화면에 뿌리기

전부 순서가 중요. 앞 단계 결과가 다음 단계 입력.

 

 

데이터 스트림

= 지금은 없지만, 나중에 올 수도 있고, 여러 번 올 수도 있는 데이터.

즉, 시간에 따라 값이 흘러오는 통로.

값이 언제 올지 모름. 몇 번 올지 모름. 안 올 수도 있음. 에러가 올 수도 있음.

 

 

기존 코드 방식

login(new Callback<User>() {
  @Override
  public void onSuccess(User user) {

    getRooms(user.id, new Callback<List<Room>>() {
      @Override
      public void onSuccess(List<Room> rooms) {

        getMessages(rooms.get(0).id, new Callback<List<Message>>() {
          @Override
          public void onSuccess(List<Message> messages) {
            show(messages);
          }

          @Override
          public void onError(Throwable e) {
            // 메시지 에러
          }
        });

      }

      @Override
      public void onError(Throwable e) {
        // 방 에러
      }
    });

  }

  @Override
  public void onError(Throwable e) {
    // 로그인 에러
  }
});

 

 

문제 상황	RxJava로 해결
흐름이 안 보임	한 줄 선언적 흐름
콜백 지옥	flatMap, Map등의 Operator 사용
에러 분산	subscribe로 한 곳에 모임
중간 개입 불가	Operator를 이용하여 Observable을 갱신
실시간 처리 어려움	스트림으로 통일

 

흐름을 코드로 그린 RxJava 예시

// 개념설명용. 흐름을 과장해서 단순화함. 
login()               // User 스트림
  .flatMap(user -> getRooms(user.id))    // Room 스트림
  .flatMap(room -> getMessages(room.id)) // Message 스트림
  .subscribe(messages -> show(messages),
             error -> showError(error))

 

이 코드를 말로 읽어보면:

로그인하고 →

그 결과로 방을 가져오고 →

그 방으로 메시지를 가져와서 →

구독한다

✔ 순서가 한 줄

✔ 에러가 한 곳

✔ 중간 개입 가능

 

 

 

RxJava 개념

 

기존 Observable에 Operator들을 체이닝하여 새로운 Observable 데이터 스트림이 생성된다.

이 때 변수로 선언하며 메모리가 할당될 수 있으나, subscribe()가 호출되기 전까지는 실행되지 않는다.

subscribe()시 Observer(onNext, onError, onComplete)가 연결되며 그 순간부터 데이터 흐름이 시작된다.

 

 

데이터 소스 Data Source

실제 데이터를 만들어내는 근원.

Observable<User> login()

 

Observable

데이터 소스를 감싼 데이터 스트림. 아직 실행 안 됨 (설계도 상태)

Observable<User> observable = login();

 

Operator

Observable을 변형해서 새 Observable을 만듦. 실행 X, 조립만 함

Observable<String> stream =
  login()
    .map(user -> user.name)
    .filter(name -> name.length() > 2);

 

Observer

데이터를 받아 처리하는 주체. 3가지 메서드로 구성.

onNext(T)
onError(Throwable)
onComplete()

 

subscribe

Observable ↔ Observer를 연결. 이 순간 실행 시작됨

observable.subscribe(
  data -> { },      // onNext
  error -> { },     // onError
  () -> { }         // onComplete
);

 

 

제일 중요한 개념은 observable과 observer를 이해하는 것이다.

 

 

 

observable과 observer에 대한 설명을 추가하면 다음과 같다.

• Cold / Hot observable

Cold Observable

예) Observable 하나, subscribe 두 번

Observable:   --1---2---3---|
Observer A:   --1---2---3---|
Observer B:   --1---2---3---|

각 subscribe마다 흐름이 새로 시작

각 Observer는 처음부터 끝까지 다 받음

 

 

Hot Observable

예) 실시간 채팅, 버튼 클릭, 센서

시간 흐름:   --1---2---3---4---5-->
Observer A:      2---3---4---5-->
Observer B:              4---5-->

Observable은 이미 흐르고 있음

구독 시점 이후 것만 받음

 

구분	Cold	Hot
subscribe 시점	처음부터 시작	중간부터
데이터 생성	매번 새로	이미 생성중
API	보통 Cold	X
실시간 이벤트	X	O

 

 

• Observer의 세 종류

onNext

한 단계가 끝날 때마다 반환. 생략가능.

 

OnError

중간에 에러 발생시 에러결과반환.

 

onComplete

모든 단계가 끝나면 결과반환.

 

 

---

https://reactivex.io/documentation/observable.html

ReactiveX - Observable