對RxJava中.repeatWhen()和.retryWhen()操作符的思考
原文作者： Daniel Lew
譯文出自： 小?子的簡書
譯者： 小?子
狀態： 完成
第一次見到.repeatWhen()和.retryWhen()這兩個操作符的時候就非常困惑了。不得不說，它們絕對是“最令人困惑彈珠圖”的有力角逐者。
然而它們都是非常有用的操作符：允許你有條件的重新訂閱已經結束的Observable。我最近研究了它們的工作原理，現在我希望嘗試著去解釋它們（因為，我也是耗費了一些精力才參透它們）。
Repeat與Retry的對比
首先，來了解一下.repeat()和.retry()之間最直觀的區別是什么？這個問題并不難：區別就在于什么樣的終止事件會觸發重訂閱。
當.repeat()接收到.onCompleted()事件后觸發重訂閱。
當.retry()接收到.onError()事件后觸發重訂閱。
然而，這種簡單的敘述尚不能令人滿意。試想如果你要實現一個延遲數秒的重訂閱該如何去做？或者想通過觀察錯誤來決定是否應該重訂閱呢？這種情況下就需要.repeatWhen()和.retryWhen()的介入了，因為它們允許你為重試提供自定義邏輯。
Notification Handler
你可以通過一個叫做notificationHandler的函數來實現重試邏輯。這是.retryWhen()的方法簽名（譯者注：方法簽名，指方法名稱、參數類型和參數數量等）：
retryWhen(Func1<? super Observable<? extends java.lang.Throwable>,? extends Observable<?>> notificationHandler)
簽名很長，甚至不能一口氣讀完。我發現它很難理解的原因是因為存在一大堆的泛型約定。
簡化后，它包括三個部分：
Func1像個工廠類，用來實現你自己的重試邏輯。輸入的是一個
Observable<Throwable>。輸出的是一個
Observable<?>。
首先，讓我們來看一下最后一部分。被返回的Observable<?>所要發送的事件決定了重訂閱是否會發生。如果發送的是onCompleted或者onError事件，將不會觸發重訂閱。相對的，如果它發送onNext事件，則觸發重訂閱（不管onNext實際上是什么事件）。這就是為什么使用了通配符作為泛型類型：這僅僅是個通知（next, error或者completed），一個很重要的通知而已。
source每次一調用onError(Throwable)，Observable<Throwable>都會被作為輸入傳入方法中。換句話說就是，它的每一次調用你都需要決定是否需要重訂閱。
當訂閱發生的時候，工廠Func1被調用，從而準備重試邏輯。那樣的話，當onError被調用后，你已經定義的重試邏輯就能夠處理它了。
這里有個例子展示了我們應該在哪些場景下訂閱source，比如，只有在Throwable是IOException的情況下請求重訂閱，否則不（重訂閱）。
source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
@Override public Observable<?> call(Observable<? extends Throwable> errors) {
return errors.flatMap(new Func1<Throwable, Observable<?>>() {
@Override public Observable<?> call(Throwable error) {
// For IOExceptions, we retry
if (error instanceof IOException) {
return Observable.just(null);
}
// For anything else, don't retry
return Observable.error(error);
}
});
}
})</pre>
由于每一個error都被flatmap過，因此我們不能通過直接調用.onNext(null)觸發重訂閱或者.onError(error)來避免重訂閱。
經驗之談
這里有一些關于.repeatWhen()和.retryWhen()的要點，我們應該牢記于心。
.repeatWhen()與.retryWhen()非常相似，只不過不再響應onCompleted作為重試條件，而是onError。因為onCompleted沒有類型，所有輸入變為Observable<Void>。每一次事件流的訂閱
notificationHandler（也就是Func1）只會調用一次。這也是講得通的，因為你有一個可觀測的Observable<Throwable>，它能夠發送任意數量的error。輸入的
Observable必須作為輸出Observable的源。你必須對Observable<Throwable>做出反應，然后基于它發送事件；你不能只返回一個通用泛型流。
換言之就是，你不能做類似的操作：
.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
@Override public Observable<?> call(Observable<? extends Throwable> errors) {
return Observable.just(null);}
})</pre>
因為它不僅不能奏效，而且還會打斷你的鏈式結構。你應該做的是，而且至少應該做的是，把輸入作為結果返回，就像這樣：
.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
@Override public Observable<?> call(Observable<? extends Throwable> errors) {
return errors;
}
})</pre>
（順便提一下，這在邏輯上與單純使用.retry()操作符的效果是一樣噠）
輸入
Observable只在終止事件發生的時候才會觸發（對于.repeatWhen()來說是onCompleted,而對于.retryWhen()來說是onError）。它不會從源中接收到任何onNext的通知，所以你不能通過觀察被發送的事件來決定重訂閱。如果你真的需要這樣做，你應該添加像.takeUntil()這樣的操作符，來攔截事件流。
使用方式
現在，假設你已大概了解了.repeatWhen()和.retryWhen()，那么你能將一些什么樣的精簡邏輯放入到notificationHandler中呢？
使用.repeatWhen() + .delay()定期輪詢數據:
source.repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
@Override public Observable<?> call(Observable<? extends Void> completed) {
return completed.delay(5, TimeUnit.SECONDS);
}
})</pre>
直到notificationHandler發送onNext()才會重訂閱到source。因為在發送onNext()之前delay了一段時間，所以優雅的實現了延遲重訂閱，從而避免了不間斷的數據輪詢。
非此即彼，使用.flatMap() + .timer()實現延遲重訂閱:
(譯者注：在RxJava 1.0.0及其之后的版本，官方已不再提倡使用.timer()操作符，因為.interval()具有同樣的功能)
source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
@Override public Observable<?> call(Observable<? extends Throwable> errors) {
return errors.flatMap(new Func1<Throwable, Observable<?>>() {
@Override public Observable<?> call(Throwable error) {
return Observable.timer(5, TimeUnit.SECONDS);
}
});
}
})</pre>
當需要與其他邏輯協同的時候，這種替代方案就變得非常有用了，比如。。。
使用.flatMap() + .timer()實現有限次數的重訂閱
source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
@Override public Observable<?> call(Observable<? extends Throwable> errors) {
return errors.zipWith(Observable.range(1, 3), new Func2<Throwable, Integer, Integer>() {
@Override public Integer call(Throwable throwable, Integer i) {
return i;
}
});
}
})</pre>
最后的結果就是每個error都與range中一個輸出配對出現，就像這樣：
zip(error1, 1) -> onNext(1) <-- Resubscribe
zip(error2, 2) -> onNext(2) <-- Resubscribe
zip(error3, 3) -> onNext(3) <-- Resubscribe
onCompleted() <-- No resubscription
因為當第四次error出現的時候，range(1,3)中的數字已經耗盡了，所以它隱式調用了onCompleted()，從而導致整個zip的結束。防止了進一步的重試。
將可變延遲策略與次數限制的重試機制結合起來
source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
@Override public Observable<?> call(Observable<? extends Throwable> errors) {
return errors.zipWith(Observable.range(1, 3), new Func2<Throwable, Integer, Integer>() {
@Override public Integer call(Throwable throwable, Integer i) {
return i;
}
}).flatMap(new Func1<Integer, Observable<? extends Long>>() {
@Override public Observable<? extends Long> call(Integer retryCount) {
return Observable.timer((long) Math.pow(5, retryCount), TimeUnit.SECONDS);
}
});
}
})</pre>
在這種用例的比較上，我認為.flatMap()+.timer()的組合比單純使用.delay()更可取，因為我們可以通過重試次數來修改延遲時間。重試三次，并且每一次的重試時間都是5 ^ retryCount，僅僅通過一些操作符的組合就幫助我們實現了指數退避算法（譯者注：可參考二進制指數退避算法）。