RxJava - Disposable Deep Dive!

1. Disposable 클래스의 역할

 

 

RxJava 공식 문서에서는 Observable과 Observer의 관계를 위 그림으로 표현한다. Observable에서 데이터를 전달 할 때는 onNext() 함수가, 더이상 전달할 값이 없을 때는 onComplete() 함수가 마지막으로 에러가 발생하면 onError()가 호출되는 방식이다. 이런 설명 방식도 조금 디테일하게 분석해보면 Observable과 Observer 사이에 Disposable 객체를 추가하는 것이 조금 더 정확할 것 같다.

 

 

Disposable 객체는 Observable에서 노출할 자원을 갖고 있고 Observer에게 이벤트로 전달하는 객체다. 그래서 RxJava 내부 소스코드를 분석해보면 첫번째 그림에 보여진 Observabe -> Observer에서 호출되는 함수는 사실 Disposable를 구현한 클래스 객체 내부에서 호출되고 있다. Observable은 Disposable을 생성하기 전까지 스트림을 대신 관리해주는 클래스고 실질적으로 값을 보내는 작업은 Disposable 내부 클래스에서 실행되고 있다.

 

2. 짧은 RxJava 코드

 

구체적으로 설명하기 위해 짧은 RxJava 코드 실행시 생성되는 객체들의 연관관계를 그려봤다. 

 

 

1. Observable.Just 는 단일 아이템을 생성하는 Observable 객체다. 이 객체를 생성하면 ObservableJust가 생성된다. 

2. doOnNext는 앞서 받은 Observable 아이템을 처리하고자 생성하는 루틴인데, 스트림을 유지하고자 ObservableDoOnEach를 만들었다. doOnNext의 내부 루틴은 DoOnEachObserver 객체에서 처리한다. 

3. doOnError도 doOnNext와 마찬가지로 스트림을 유지하고자 ObservableDoOnEach를 만들었다. doOnErro 내부 루틴은 DoOnEachObserver에서 처리한다. 

4. subscribe()가 호출되면 아래 스트림부터 최상단 스트림까지 차례로 구독 관계가 형성된다. ObservableDoOnEach -> ObservableDoOnEach -> ObservableJust 순서로 subscribe가 재귀로 호출되면서 Observer 간의 구독 관계가 완성된다

5. 최상단 ObservableJust는 값 1을 발행하는데 이 이벤트는 ScalarDisposable 에서 담당한다. ObservableJust는 DoOnEachObserver 내부 onSubscribe 함수를 호출해서 스트림 간에 down/upstream 을 구축한다.

6. ScalarDisposable 내부에서는 DoOnEachObserver 내부 onNext 함수를 호출해서 값 1을 전달한다. 

 

3. ScalarDisposable

 

ScalarDisposable 에서 값을 전달하는 부분은 Runnable로 동작하게끔 구현되있다. Observable.just 형태의 스트림을 구독하면 매번 새로운 쓰레드가 생성되서 실행된다.

 

 

4. Memory Leak 가능성

 

다른 Disposable 를 구현한 클래스를 찾아보면 Observable의 역할에 따라서 Runnable인 경우도 있고 Scheduler로 돌리는 경우도 있다. Intervar 처럼 긴 시간 돌리는 작업이면 따로 dispose() 함수를 호출하지 않는 이상 쓰레드가 종료되지 않고 계속 실행된다. 이런 코드가 증가하게 되면 불필요한 쓰레드 개수가 늘어나 Memory Leak이 발생할 소지가 있다. RxJava()를 사용할 때 CompositeDisposable() 객체를 활용해서 dispose() 시키라는 이유가 여기에 있다.