[RxJava] 6.Scheduler

Scheduler

1. Schedulers 개요

Observable 연산자 체인에 멀티스레딩을 적용하고 싶다면, 특정 스케줄러를 사용해서 연산자(또는 특정 Observable)를 실행하면 된다.

ReactiveX의 일부 Observable 연산자는 사용할 스케줄러를 파라미터로 전달받기도 하는데, 이 연산자들은 자신이 처리할 연산의 일부 또는 전체를 전달된 스케줄러 내에서 실행한다.

기본적으로, Observable과 연산자 체인은 이처럼 스케줄러를 통해 동작하고 Subscribe 메서드가 호출되는 스레드를 사용해서 옵서버에게 알림을 보낸다. SubscribeOn 연산자는 다른 스케줄러를 지정해서 Observable이 처리해야 할 연산자들을 실행시킨다. 그리고, ObserveOn 연산자는 Observable이 옵서버에게 알림을 보낼 때 사용할 스케줄러를 명시한다.

 

아래 그림이 보여주듯, SubscribeOn 연산자는 Observable이 연산을 위해 사용할 스레드를 지정하며, 연산자 체인 중 아무 곳에서 호출해도 문제되지 않는다. 하지만, ObserveOn 연산자는 연산자 체인 중 Observable이 사용할 스레드가 호출 체인 중 어느 시점에서 할당되는지에 따라 그 후에 호출되는 연산자는 영향을 받는다. 그렇기 때문에, 어쩌면 여러분은 특정 연산자를 별도의 스레드에서 실행 시키기 위해 연산자 체인 중 한 군데 이상에서 ObserveOn을 호출하게 될 것이다.

 

SubscribeOn

SubscribeOn은 구독(subscribe)에서 사용할 스레드를 지정한다.

도중 ObserveOn이 호출되어도 SubscribeOn의 스레드 지정에는 영향을 끼치지 않는다.

 

ObserveOn

ObserveOn은 Observable이 다음처리를 진행할 때 사용할 스레드를 지정한다.

ObserveOn이 선언된 후 처리가 진행뒤 다른 ObserveOn이 선언 시 다른 ObserveOn에서 선언한 스레드로 변경되어 이후 처리를 진행한다.

 

2. Schedulers 종류

▶ Schedulers.newThread()

각 활성화 된 Observable을 따라 NewThread를 생성한다. 작업에 필요한 새로운 Thread를 생성하는 건 상당한 비용이 드는 작업이고 성능부하를 줄 수 있기 때문에 특별한 경우가 아니라면 사용을 지양하는 것을 권장한다.

 

▶ Schedulers.computation()

Computation 스케줄러도 동일하게 자바의 Thread pool을 이용하며, 특징은 디바이스의 CPU 코어 갯수만큼으로 스레드 수가 한정된다는 점이다(듀얼 코어라면 두 개, 쿼드라면 4개). 이는 경우에 따라 한정된 스레드들이 전부 바쁜 상태에 있는 경우 다음 작업을 진행하기 위해 대기를 해야 하는 상황이 생길 수 있다는 것이다. 참조한 포스트에서는 적은 양의 연산, 그리고 이를 빠르게 반영하고 싶은 경우 사용하는 것을 권장하고 있다.

 

▶ Schedulers.from(executor)

커스텀한 IO Scheduler라고도 말할 수 있으며, Thread pool의 사이즈를 직접 명시해 설정이 가능하다. IO Scheduler로 처리가 어려운, 다수의 입력을 처리하고자 할 때 사용하면 유용하다.

사용법은 executor를 우선 생성하고, scheduler를 Schedulers.from(executor)로 생성해 활용한다.

val executor = Executors.newFixedThreadPool(10) 

val pooledScheduler = Schedulers.from(executor)

 

▶ Schedulers.immediate()

어떤 프로세스가 동작하고 있는지 여부에 상관없이 작업을 즉시 수행하는 스케줄러로 RxJava 2에서 삭제되었다.

 

▶ Schedulers.io()

IO 스케줄러는 네이밍대로 네트워크 관련 작업이나 파일 시스템과의 상호작용시 사용하는 스케줄러이다.

자바의 Thread pool(작업을 기다리고 재사용되는 Worker thread 그룹)을 통해 구동되며, 기존 Thread가 재사용되지 못하는 조건에 있는 경우 새로운 Thread를 생성해 작업을 수행한다.

 

이 특징 때문에 경우에 따라 엄청난 양의 Thread를 만들 수 있고, 이로 인해 성능저하가 생길 수 있다는 단점이 있다.

 (Thread는 60초 동안 사용되지 않았을 경우에 제거됨)

 

▶ Schedulers.trampoline()

트램폴린 Scheduler는 코드를 현재 Thread에서 수행한다. 고로 만약 Main Thread에서 코드를 돌리고 있었다면 코드블럭은 Main Thread의 Queue에 추가된다. Thread를 블록한다는 부분에서 Immediate Scheduler와 유사하나 차이점은 Thread에서 진행중인 작업이 완전히 끝날 때까지 기다린다는 점이다. 여러 개의 Observable로부터 들어오는 흐름을 제어할 때 사용하는 것을 권장한다. (들어온 데이터를 순차적으로 처리하는 것을 보장)

 

▶ AndroidSchedulers.mainThread() 

rxAndroid library를 통해 제공되는 Scheduler로 특정 동작을 UI thread에서 수행하도록 한다. 작업이 끝난 결과를 UI에 반영하고자 할 때 지정해주는 방식으로 사용한다.

 

▶ Single

Observable의 갯수가 몇개가 되는지에 상관없이 단 하나의 Thread에서 작업을 수행하는 스케줄러이다. 

 

▶ HandlerScheduler.from(handler) 

특정 핸들러 handler에 의존하여 동작합니다.

 

 

 

원문: Scheduler

http://reactivex.io/documentation/scheduler.html

 

RxJava — Schedulers

https://medium.com/android-news/rxjava-schedulers-what-when-and-how-to-use-it-6cfc27293add