Observable
RxJava의 가장 핵심적인 요소는 Observable입니다. Observable은 데이터 흐름에 맞게 알림을 보내 Observer가 데이터를 사용할 수 있도록 합니다. 즉, Observable을 이용해 데이터를 회수하고 변환하는 메커니즘을 정의한고, Observer는 이를 구독해 데이터가 준비되면 이에 반응합니다.
Observer 패턴 글입니다.
Observable은 Collections(List, ArrayList, etc)를 사용할 때 같은 방식으로 비동기 이벤트 스트림을 처리할 수 있습니다. 다만 Collections의 Iterable이 Push 방식이라면, Observable은 Iterable의 Pull 버전입니다. Iterable은 Consumer(데이터를 소비하는 곳)가 값을 Pull한 후 값이 도착할 때까지 기다리며 Thread를 차단한다면 Observable은 Thread를 차단하지 않고 값이 사용가능하면 Consumer에게 값을 Push합니다.
정리하면
-
Observable이 데이터 스트림을 처리하고 완료되면 데이터를 발행합니다.
-
데이터를 발행할 때마다 구독하고 있는 모든 Observe가 알림을 받습니다.
-
Observer는 수신한 데이터를 가지고 어떠한 일을 합니다.
이러한 패턴을 사용하게 되면 동시성이 보장될 수 밖에없습니다. Observable이 데이터를 발행할 때까지 계속 기다릴 필요없고 단지 Observer가 데이터가 발행될 때 동작만 제대로 하면되기 때문입니다.
Emitter 인터페이스
Reactive X에서는 바로 Emitter라는 인터페이스가 Push 역할을 하게 됩니다. Observable은 Emitter에 포함되어 있는 3가지 이벤트를 활용하여 감시자에게 무언가 알릴 수 있습니다.
onNext()
- 데이터가 하나 발행됐음을 알리는 이벤트
- 연속된 데이터의 경우, 데이터가 하나씩 차례대로 onNext()로 발행됨
onComplete()
- 데이터 발행이 모두 완료되었음을 알리는 이벤트
- onComplete()가 호출된 이후에는 더 이상 onNext() 호출이 안됩니다.
onError()
- 오류가 발생했음을 알리는 이벤트
- 마찬가지로 onError()가 호출된 이후에는 더 이상 onNext() 호출이 안됩니다.
중요한 점
-
Observable에서 데이터, 오류 등을 발행할 때 null 발행은 안됩니다.
-
항상 onComplete()혹은 onError() 둘 중 하나로만 데이터 발행이 종료되어야합니다.
