RxJava郵編優先

Question

這是問題所在。我有一些異步操作，其結果彙總成一個進一步處理。但是，並非所有的操作都是相同的，並且錯誤處理根據哪個操作失敗而不同。如果A失敗，我們需要結束處理，但如果B或C失敗，我們將繼續正常處理其他處理。如果A失敗，我們需要結束處理，但是如果B或C失敗，我們繼續正常處理其他處理。

目前，我們使用倒計數鎖存器和大量的狀態管理實現了接近一百行代碼。我想將它移到基於RxJava的實現中。我的第一個想法是嘗試Observable.zip算子，但它將所有觀測值視爲平等，在這種情況下不是這樣。我的另一個想法是把電話連接起來，這是有效的，但這意味着所有的操作不會同時開始，導致更長的總體時間。

任何人都可以指導我如何做到這一點？

Answer 1

使用.onErrorResumeNext：

Observable<T> a, b, c; 
Observable.zip(
    a, 
    b.onErrorResumeNext(t -> Observable.just(null)), 
    c.onErrorResumeNext(t -> Observable.just(null)), 
    (x, y, z) -> <your aggregation>) 
... 

代表與null的誤碼觀測是由你。你也可以使用Optional：

Observable.zip(
    a, 
    b.map(x -> Optional.of(x)) 
    .onErrorResumeNext(t -> Observable.just(Optional.empty())), 
    c.map(x -> Optional.of(x)) 
    .onErrorResumeNext(t -> Observable.just(Optional.empty())), 
    (x, y, z) -> <your aggregation>) 

如果b和c對應於外部服務調用，你可能還需要忽略他們，如果他們花費太長時間用b.timeout(5, TimeUnit.SECONDS)更換b。簡潔呃！