Android RxJava - Scheduler

Scheduler

Scheduler는 RxJava에서 어떤 쓰레드에서 실행될지 결정하는 역할을 합니다. 단순히 Observable를 만든다고 비동기 실행이 되는 것이 아닌, Scheduler를 적절히 사용해야 비동기로 실행할 수 있습니다.

 

Scheduler를 사용하지 않은 경우

Observable를 만들지만 내부적으로 같은 쓰레드를 사용하기 때문에 순차적으로 실행된다.

fun schedulerTest1() {
    Observable.create<Int> {
        Thread.sleep(1500L)
        it.onNext(100)
    }.subscribe {
        println("subscribe : $it")
    }

    println("Finish")
    Thread.sleep(3000L)
}

subscribe : 100
Finish

 

Scheduler를 사용한 경우

Observable은 RxJava에서 사용하는 쓰레드에서 실행되면서 비동기로 작업이 진행된다.

fun schedulerTest2() {
    Observable.create<Int> {
        Thread.sleep(1500L)
        it.onNext(100)
    }.subscribeOn(
        Schedulers.io()
    ).subscribe {
        println("subscribe : $it")
    }

    println("Finish")
    Thread.sleep(3000L)
}

Finish
subscribe : 100

 

Scheduler 변환

subscribeOn, observeOn으로 쉽게 변환할 수 있다.

• subscribeOn : 발행하는 스트림의 쓰레드를 정할 수 있습니다. 지정하지 않으면 Observable를 만든 쓰레드에서 실행되고 다른 쓰레드로 지정하고 싶은 경우 subscribeOn을 사용하면 된다. (subscribeOn는 호출하는 순서에 상관없이 첫번째로 호출한 subscribeOn으로 결정된다.)
• observeOn : Consumer가 작업하는 쓰레드를 정할 수 있습니다. 지정하지 않으면 발행하는 스트림과 같은 쓰레드에서 실행되고 다른 쓰레드로 지정하고 싶은 경우 observeOn을 사용하면 됩니다. observeOn은 호출하는 순서에 따라 영향을 줍니다. doOnXXX 함수를 사용할 경우 가장 최근에 실행한 observeOn의 쓰레드에서 실행됩니다.

 

Schedulers.io()

가장 많이 사용하는 Scheduler입니다. 일반적으로 io 작업을 처리할 때 사용하며 쓰레드 풀을 가지고 있습니다. 내부적으로 쓰레드 하나로 시작하여 새로운 쓰레드가 필요할 때(새로운 작업을 진행할 때 여유분의 쓰레드가 없는 경우 생성) 추가로 생성하여 사용합니다. 쓰레드 생성에 제한이 없기 때문에 무한정 생성될 수 있으며 성능에 영향을 줄 수 있습니다.

 

Schedulers.computation()

io와 비슷하지만 쓰레드 풀의 수가 제한(쓰레드의 수와 같습니다.)되어있습니다. 한번에 많은 작업이 들어오면 io는 새로운 쓰레드를 만들어서 처리하지만, computation은 작업을 기다렸다가 쓰레드를 사용할 수 있을 때 처리합니다. 쓰레드의 수가 제한되어 있기 때문에 실행 시간이 짧은 작업을 처리하는 것이 효율적입니다.

 

Schedulers.trampoline()

trampoline을 실행하는 쓰레드의 크기가 무한한 큐를 만들고 작업을 순차적으로 실행합니다. 작업을 완료해야 다음 작업을 실행하기 때문에 순서가 중요할 때 사용합니다.

 

Schedulers.single()

Rx에서 미리 만들어둔 쓰레드에서 작업을 진행합니다. 여러개의 작업이 들어와도 하나의 쓰레드에서 실행됩니다. Rx는 비동기 프로그래밍을 지향하기 때문에 사용률이 낮습니다.

 

Executors

Rx에서 지원하는 Schedulers 대신 Java에서 기본적으로 지원하는 Executors를 통해 커스텀화 할 수 있습니다.

fun executor() {
    val executors = Executors.newFixedThreadPool(10)
    Observable.create<Int> {

    }.subscribeOn(
        Schedulers.from(executors)
    )
}

 

AndroidSchedulers.mainThread()

Android의 MainThread에서 작업을 처리합니다. UI 업데이트 작업을 하는 것이 좋습니다.