[譯]RxJava 的全面介紹：Observable 型別、背壓、錯誤處理

RxJava 安卓開發 · 發表 2019-01-28 10:28:48

摘要： RxJava 是一個不斷更新的工具庫，適用於除Android 以外的許多平臺的開發人員（如：RxSwift ）。RxJava 最大的優勢是以不使用回撥的方式處理非同步操作。相反，Observables 和Observers...

RxJava 是一個不斷更新的工具庫，適用於除Android 以外的許多平臺的開發人員（如：RxSwift ）。RxJava 最大的優勢是以不使用回撥的方式處理非同步操作 。

相反，Observables 和Observers 結合使用來發射資料（一次或多次），並且還可以通過各自包含的方法來處理每次資料發射時要做的事情。

什麼是 Observable 和 Observer

val myObservable = Observable.just(1, 2, 3)
 
val myObserver = myObservable.subscribe { receivedNumber ->
Log.d(LOG_TAG, "Received Number $receivedNumber")
}
複製程式碼

Observable–發射資料流並與接受改資料的Observer 一起工作的物件。Observer –訂閱一個Observable 讓它開始發射資料，並處理接受資料時要做的事情。

在上面的例子中，Observable.just(1, 2, 3) 將按照順序發射整數 1, 2, 3。Observer 一旦訂閱Observable ，將以相同的順序接受這些數字。

Received Number 1
Received Number 2
Received Number 3
複製程式碼

Observable 的生命週期

Observable 有兩個重要的方法，來處理傳入的資料。

onNext -每當發射新資料時呼叫，正如你在上面示例中的lambda 函式看到的一樣（在subscribe 之後呼叫）。
onComplete - 在沒有更多資料發射時呼叫。顧名思義，資料流完全發射完畢。

不同型別的 Observable

最基本的Observable 會發射連續的資料流，直到呼叫onComplete 為止，這並不總是你想要的。你可能想發射單個值 ，或者發射一個無法接受該值的值 ，亦或是在執行沒有返回值的非同步任務後呼叫其他函式 。

val mySingle = Single.just(1)
val singleObserver = mySingle.subscribe { data ->
Log.d(LOG_TAG, "Received $data")
}
 
val myMaybe = Maybe.empty<Int>()
val maybeObserver = myMaybe
.defaultIfEmpty(1)
.subscribe { data ->
Log.d(LOG_TAG, "Received $data")
}
 
val myCompletable = Completable.complete()
val completableObserver = myCompletable
.subscribe {
Log.d(LOG_TAG, "Task completed")
}
複製程式碼

Single僅發射一個值。onNext 只調用一次，並且onComplete 將立即被呼叫。

Mayble發射一個或零個值，當發射零個值時，將跳過onNext 並立即呼叫onComplete 。可以使用defalutIfEmpyty 函式發射預設值。

Completable不發射任何值，你可以像沒有返回值的回撥一樣訂閱它。

Flowables 和背壓

還有一種型別的Observable ，它就是Flowable 。和Observable 一樣，Flowable 也發射連續的資料流，直到完成為止。但有一個關鍵的區別：

想象一下，上游Observable 資料發射的速度，大於下游Observer 處理資料的速度，這就是背壓。在大部分情況下，將會導致錯誤發生。Flowable 是一種包含背壓策略的Observable ，具有當背壓發生時，如何處理資料的能力。

val myFlowable = Observable.range(1, 100000).toFlowable(BackpressureStrategy.DROP)
val flowableObserver = myFlowable.subscribe {data ->
Log.d(LOG_TAG, "Received $data")
}
複製程式碼

有 5 種不同的背壓策略，我們需要了解下：

Buffer -在記憶體中快取事件，直到Observer 可以處理它們。預設情況下，在錯誤發生之前，緩衝區的大小為 128 個items 。可以修改緩衝區的大小，但請注意，這將會帶來效能上的開銷。
Drop - 丟棄Observer 無法處理的事件。
Latest - 僅保留最新發射的值，直到Observer 可以使用它並丟棄其它的值。
Error - 如果發生背壓，將丟擲異常。
Missing - 缺乏背壓策咯，如果你想在客戶端處理背壓，你可以使用它（因為背壓策咯是由 Observable 建立的）。未能說出策咯會在背壓下丟擲異常。

Observable vs Flowable

已知的是，當你明確知道發射的資料，不會導致發生背壓。你應當使用Observable ，而不是Flowable 。老實說，我還沒有發現使用Flowable 的場景，也許Flowables 將使用額外的記憶體？

錯誤處理

沒有任何程式碼可以避免錯誤，你已經知道，如果不處理背壓將導致錯誤發生。最重要的是，在Observer 的subscribe 方法中，由自己的程式碼發生的異常，都將被視為由Observer 應處理的錯誤。

高興的是，RxJava 包含了幾種處理這些錯誤的方法。

doOnError - 當錯誤發生，只需執行該方法

val observer = myObservable
.doOnError { Log.e(LOG_TAG, "ErrorOccurred") }
.subscribe()
複製程式碼

onErrorReturnItem - 如果發生錯誤，返回一個預設值

val observer = myObservable
.onErrorReturnItem(0)
.subscribe()
}
複製程式碼

onErrorReturn - 和onErrorReturnItem 一樣，但接受一個返回所需資料型別的函式（對於動態預設值）

val observer = myObservable
.onErrorReturn{ throwable -> throwable.message}
.subscribe()
複製程式碼

onErrorResumeNext - 如果發生錯誤，則返回一個預設資料流，也可以為動態資料採取功能。

val observer = myObservable
.onErrorResumeNext(Observable.just(2, 4, 6))
.subscribe()
複製程式碼

retry - 如果發生錯誤，嘗試重新訂閱Observable ，你可以設定最多的重試次數，或者設定空值以便無限次重試。

val observer = myObservable
.retry(3)
.subscribe()
複製程式碼

你也可以通過布林值，來實現重試條件 。

val observer = myObservable
.retry{ integer, throwable -> integer > 0 }
.subscribe()
複製程式碼

如果沒有使用上述的操作符，在錯誤發生時，將導致程式崩潰。

結語

到目前為止，此部落格的大部分內容都是關於Firebase的。有充分的理由相信，Google 通過強大的雲資料庫、機器學習和雲服務，將推動Firebase 成為最優秀的雲解決方案之一，並也為伺服器開發打開了新的世界。

雖然我會回到Firebase 上，但從現在開始，我將潛心學習RxJava 並深入瞭解它的細節。因為它還有更多的閃光點待我去學習：修復回撥地獄、提高行能、執行緒排程.....