[RxJava] RxJava의 핵심 스케줄러 - 1 :뉴 스레드 스케줄러 , 계산 스케줄러, IO 스케줄러

스케줄러는 RxJava의 핵심이라고 말할 수도 있을 만큼 중요한 요소인데요.

어떤 프로그램의 세부 일정을 주관하는 관리자로 생각하면 이해하기 쉽습니다!

 

이제 RxJava의 종류에 대해서 알아보겠습니다!

[ RxJava 2.x 기준]

- 뉴 스레드 스케줄러 = newThread()

- 싱글 스레드 스케줄러 = single()

- 계산 스케줄러 = computation()

- IO 스케줄러 = io()

- 트램펄린 스케줄러 = trampoline()

 

이렇게 다섯가지로 구분할 수 있어요. 이제 각 스케줄러를 보다 구체적으로 알아볼게요!

 

아! 들어가기 전에 함께 자주 쓰이는 메소드 몇 개를 간단하게 알아볼게요. (다음 글들에서 자세히 다룰 예정입니다!)

- doOnNext(): Observable에서 onNext 이벤트가 발생하면 실행됩니다.

- subscribeOn(): 구독자가 Observable에 subscribe() 함수를 호출하여 구독할 때 실행되는 스레드를 지정합니다.

- observeOn(): Observable에서 생성한 데이터 흐름이 여기저기 함수를 거치며 처리될 때 동작이 어느 스레드에서 일어나는지 지정할 수 있습니다.

 

1. 뉴 스레드 스케줄러

이름 그대로 새로운 스레드를 생성하는 것입니다. 새로운 스레드를 만들어 어떤 동작을 실행하고 싶을 때 Schedulers.newThread()를 인자로 넣어주면 됩니다.

 

코드로 한 번 보겠습니다.

val args = arrayOf("1","2")

Observable.fromArray(args)
.doOnNext(it->"value1: $it")
.subscribeOn(Schedulers.newThread())
.subscribe(Log::i)

// result  (메인 스레드가 아닙니다)
// 1
// 2

지금과 같은 동작을 수행할 때마다 새로운 스레드에서 작업이 수행됩니다!

 

하지만 이 뉴 스레드 스케줄러는 적극적으로 추천하는 방법이 아니에요. 왜냐, 뉴 스레드 스케줄러보다 활용도가 높은 계산 스케줄러, IO 스케줄러등이 있기 때문이죠

 

2. 계산 스케줄러

계산 스케줄러는 CPU에 대응하는 계산용스케줄러입니다. 계산 작업을 할 때는 대기 시간 없이 빠르게 결과를 도출하는 것이 중요합니다.

내부적으로 스레드 풀을 생성하며 스레드 개수는 기본적으로 프로세서 개수와 동일합니다.

실행 방법은 subscribeOn()의 인자로 Schedulers.computation()을 넣어주면 됩니다.

 

source.map(it-> it+1)
.subscribeOn(Schedulers.computation())
.subscribe(println(it))

ㅇ이런식으로요!  interval() 함수도 적용시킬 수 있으니 한번 코딩해보시는걸 추천드립니다!

 

3. IO 스케줄러

IO 스케줄러는 계산 스케줄러와는 다르게 네트워크상의 요청을 처리하거나 각종 입 · 출력 작업을 위한 스케줄러입니다. 계산 스케줄러와 다른 점은 기본으로 생성되는 스레드 개수가 다르다는 것이죠!

 

즉, 계산 스케줄러는 cPU 개수만큼 스레드를 생성하지만 IO 스케줄러는 피룡할 때마다 스레드를 계속 생성합니다.

입 · 출력 작업은 비동기로 실행되지만 결과를 얻기까지 대기 시간이 깁니다.

- 계산 스케줄러: 일반적인 계산작업.

- IO 스케줄러: 네트워크상의 요청, 파일 입출력, DB 쿼리 등.

 

실행코드는 다음과 같습니다.

source.map(it.getAbsolutePath())
.subscribeOn(Schedulers.io())
.subscribe(println(it))

subscribeOn() 의 인자에 Schedulers.io()를 넣어주면 됩니다.