[RxJava3] subscribeOn과 observeOn

RxJava 코드를 보는데, subscribeOn과 observeOn이 함께 쓰이는 것을 보고

둘이 무슨 차이이고 어떤 동작을 하는 것인지 궁금하여 한번 정리해 보았다.

 

 

subscribeOn

Observable 이 데이터 흐름을 발생시키고 연산하는 스레드를 직접 지정할 수 있다.

observeOn

Observable 이 Observer 에게 알림을 보내는 스레드를 지정할 수 있다.

이렇게 얘기하면 잘 와닿지 않을 것이다. 사용 시, 둘의 차이점이 어떤지 더 자세히 알아보자.

// # 1
myService.getUsers()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .flatMap(Observable::fromIterable)
    .filter(User::isMember)
    .map(this::saveToCache)
    .toList()
    .subscribe(View::showUser);
    
    
// # 2
myService.getUsers()
    .subscribeOn(Schedulers.io())
    .flatMap(Observable::fromIterable)
    .filter(User::isMember)
    .map(this::saveToCache)
    .toList()
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(View::showUser);

각 1, 2번에 대해서 차이를 보자면, subscribeOn 은 observeOn 의 위치 전까지의 연산을 io Schedulers 에서 수행하고, observeOn 의 경우는 본인 하위 연산에 대해 mainThread 에서 수행한다.

따라서, 1번의 경우는 getUser만 io 스케줄러에서 수행하고, 나머지는 mainThread 에서 수행된다.

2번의 경우는 getUsers 부터 toList 까지 io 스케줄러에서 수행되고, 나머지 subscribe 는 mainThread에서 수행된다.

안드로이드 lifecycle 상 메인스레드에서 많은 연산을 할 경우 앱의 자원 낭비가 일어날 수 있기 때문에, 메인 스레드에서는 ui 만 변경하는 작업을 하는 것이 좋다.

 

 

따라서, 위 예시의 더욱 적당한 코드는 아래처럼 getUsers 연산은 io 로 가져오고 이후의 flatMap, filter 작업은 연산에 적합한 Schedulers.computation() 으로 처리한다. 해당 작업이 끝난 후에는 observeOn 으로 subscribe 는 다시 mainThread 로 하여 UI 에 반영될 수 있도록 한다.

myService.getUsers()
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.computation()) 
    .flatMap(Observable::fromIterable)
    .filter(User::isMember)
    .map(this::saveToCache)
    .toList()
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(View::showUser);

 

 

간략하게 정리하면 subscribeOn은 obeserver에 의해 subscribe 되었을 때, source 가 데이터를 다음 스트림으로 전달하는 스케줄러는 지정한다. observeOn 은 본인 이후 수행되는 스트림의 액션을 수행하는 스케줄러를 지정한다. 따라서, 어떤 순서로 호출하느냐에 따라 어떤 스레드에서 처리되는지의 흐름을 정할 수 있는 것이다.

추가로, subscribeOn은 최초 한번 호출된 것만 유효하며 뒤에 중복 호출이 될 경우는 무시된다. 반면 observeOn의 경우 횟수 제한 없이 호출이 가능하다.

• I/O 스케줄러
• 운영체제에서 저장소에 파일의 입/출력을 관리하거나, 네트워킹 작업이 있을 경우 등 블로킹이 발생할 수 있는 곳에서 비동기적으로 작업을 처리하기 위해 주로 사용되는 방식이다. 스레드 풀을 사용해 새로운 스레드가 필요할 때마다 계속 생성하되, 이전에 생성된 스레드가 존재한다면 이를 재사용한다.

 

 

참고 링크

• subscribeOn, observeOn: https://choi-dev.tistory.com/m/139
• IO 스케줄러: https://velog.io/@haero_kim/RxJava-Scheduler-설정하기