RxJava 노트

RxJava 노트

just, fromCallable, defer 차이

// Sample

fun fromCallableTemp() { Timber.d("fromCallableTemp() | start | ${System.currentTimeMillis()}") Observable.just(getDataWithSleep()) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribeWith(object : DisposableObserver() { override fun onComplete() { Timber.d("fromCallableTemp | onComplete() called.") } override fun onNext(t: String) { Timber.d("fromCallableTemp | onNext() | $t") } override fun onError(e: Throwable) { Timber.d("fromCallableTemp | onError: %s", e) } }) Timber.d("fromCallableTemp() | end | ${System.currentTimeMillis()}") } fun getDataWithSleep(): String { try { Thread.sleep(3000) } catch (e: InterruptedException) { e.printStackTrace() } Timber.d("getDataWithSleep() | ${Thread.currentThread().name} ") return "Data With Sleep" }

just

위와 같은 코드에서 just는 메인쓰레드 블락을 야기한다. 그래서 무거운 로직은 쓰면 안된다.

Scheduling을 했지만 main 에서 돈다.

Rx 자바 쓰레드 체인징이 안먹히는 것을 확인할 수 있다.

fromCallable, defer

쓰레드 블락을 하지 않고 스트림 생성을 지연한다.

subscribeOn의 영향을 받는 것을 확인할 수 있다.

Thread Changing

private fun test() { Observable.fromCallable { heavyJob() } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .doOnNext { Timber.d("currentThread1: %s", Thread.currentThread()) } .observeOn(Schedulers.io()) .doOnNext { Timber.d("currentThread2: %s", Thread.currentThread()) } .onErrorReturnItem(1) .observeOn(AndroidSchedulers.mainThread()) .subscribe() } private fun heavyJob(): Int { Timber.d("heavyJob Start") Thread.sleep(5000L) Timber.d("currentThread in heavyJob: %s", Thread.currentThread()) Timber.d("heavyJob End") return 1 }

결과:

doOnNext는 데이터를 발행할 때 작동되지만, 이처럼 체이닝 위치에 따라 쓰레드 스케쥴링이 가능하다는 것을 알 수 있다.

에러핸들링

RxJava는 에러 핸들링이 try-catch로 불가능하지는 않다.

try catch로 해도 아이템 발행은 할 수 있게 가능하다. 하지만 부자연스럽다.

RxJava는 onError 연산자들을 지원하고 좀 더 자연스럽고 예외처리 조건을 다양하게 줄 수 있기 때문에 onError연산자들을 쓰는게 좋다고 본다. try catch는 단순 에러를 내뱉는 반면 onError~ 연산자를 쓰면 다양한 방식으로 처리할 수 있기 때문이다.

또한 subscribe 시에 주의할점은 onError 연산자, try catch를 써서 예외처리를 하지 않을 때 exception이 발생한다면

아래와 같은 메세지를 내뱉으면서 앱이 죽을 것이다. 그래서 단순 subscribe()를 호출할 것이라면 예외처리를 반드시 하자.

The exception was not handled due to missing onError handler in the subscribe() method call. Further reading: https://github.com/ReactiveX/RxJava/wiki/Error-Handling | java.lang.IllegalArgumentException:

또한 유의할점은 onError~ 연산자는 다운스트림을 패스하지 않고 바로 조건대로 아이템을 발행한다.

private fun test() { Observable.fromCallable { heavyJob() } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .doOnNext { Timber.d("currentThread1: %s", Thread.currentThread()) } .observeOn(Schedulers.io()) .doOnNext { throw IllegalArgumentException("") } .onErrorReturnItem(99999) .onErrorReturn { if (it is IllegalArgumentException) { 333 } else { 444 } } .observeOn(AndroidSchedulers.mainThread()) .subscribe({ Timber.d("result: %s", it) }) }

위와 같이 onError연산자는 바로 스트림을 종료시키고 해당 조건에 따라 방출하기 때문에 다운스트림이 먹히지 않는다.위의 코드에서는 99999를 발행한다.

반대로 onErrorReturn이 업스트림에 있으면 그 방식대로 발행하며 해당 스트림은 종료된다.

from http://dunkey2615.tistory.com/245 by ccl(A) rewrite - 2021-10-28 11:27:12