RxJava subscribeOn dan Thread

Operator subscribeOn memiliki definisi yang sederhana tapi memiliki perilaku yang kadang tidak disadari pengguna. Sebagai contoh, lihat snippet di bawah:

Observable.create<Int> {
    Thread(Runnable {
        it.onNext(1)
        it.onComplete()
    }, "Thread A").start()
}
.subscribeOn(Schedulers.newThread())
.subscribe {
    println("Subscriber on ${Thread.currentThread()}")
}

Di thread apakah subscriber berada?

a. new thread,

Subscriber on RxNewThreadScheduler-1

b. Thread A

Subscriber on Thread A

Operator subscribeOn

Mengutip reactive.io, subscribeOn menentukan operasi Observable dilakukan menggunakan scheduler apa. Scheduler bertugas menentukan di thread apa operasi dilakukan. Singkatnya, dengan subscribeOn, kita dapat menentukan di thread apa operasi observable dilakukan.

Tanpa subsribeOn, maka operasi akan dilakukan di thread di apa observable di subscribe (main thread pada contoh).

Maka snippet di bawah

Observable.create<Int> {
        it.onNext(1)
        it.onComplete()
    }
        .subscribeOn(Schedulers.newThread())
        .subscribe {
            println("Subscriber on ${Thread.currentThread().name}")
        }

akan memberikan output:

Subscriber on RxNewThreadScheduler-1

Subject dan subscribeOn

Subject adalah proxy yang dapat berlaku sebagai observer dan observable. Sederhananya subject mengekspos fungsi-fungsi observer sehingga dapat diakses.

val subject = PublishSubject.create<Int>()

subject
        .subscribeOn(Schedulers.newThread())
        .subscribe {
            println("Subscriber on ${Thread.currentThread().name}")
        }

subject.onNext(1)
subject.onComplete()

Di thread apa subscriber berada untuk snippet di atas? Bukan new thread, melainkan main thread.

Mengapa? Karena subject mengekspos onNext dan fungsi tersebut dieksekusi di main thread. Dengan begitu, subscribeOn kehilangan manfaatnya, yaitu mengeksekusi operasi observable di new thread.

Penutup

Mari kembali ke snippet di awal artikel. Walaupun observable di subscribe menggunakan scheduler new thread, fungsi onNext dan onComplete dipanggil di dalam Thread A persis subject dan subscribeOn di atas.

Bila digambarkan sebagai diagram, maka snippet di awal akan seperti ini:

Ekstra

Hindari mengandalkan subscribeOn untuk menentukan di thread apa subscriber beroperasi. observeOn adalah operator yang tepat untuk menentukan di thread apa subscriber beroperasi.

Bila tidak ditentukan, maka subscriber akan beroperasi di thread yang sama dengan observable beroperasi.

Referensi