如何在RxJava出錯後繼續流式傳輸項目？

我是RxJava新手，我有以下問題。說我有項目的序列和項目傳播錯誤，我想忽略它，並繼續處理其他項目。如何在RxJava出錯後繼續流式傳輸項目？

我有下面的代碼片段：

Observable.from(Arrays.asList("1", "2", "3")) 
      .map(x -> { 
       if (x.equals("2")) { 
        throw new NullPointerException(); 
       } 
       return x + "-"; 
      }) 
      .onExceptionResumeNext(Observable.empty()) 
      .subscribe(System.out::println);

我越來越：1-

但我想：1，3

我怎麼能這樣做？

來源

2016-11-02 corvax

的可能的複製【如何忽略錯誤並繼續無限流？（http://stackoverflow.com/questions/28969995/how-to-ignore-error-and-continue-infinite-stream） – Ivan

訣竅是將值以某種方式轉換爲新的可觀察值和平面圖，如下例所示。現在，flatMap中的每個值都可以引發異常並按值處理值。因爲flatMap中的子流只包含一個元素，所以在onError之後observable將被關閉並不重要。我使用RxJava2作爲測試環境。

@Test 
public void name() throws Exception { 
    Observable<String> stringObservable = Observable.fromArray("1", "2", "3") 
      .flatMap(x -> { 
       return Observable.defer(() -> { 
        try { 
         if (x.equals("2")) { 
          throw new NullPointerException(); 
         } 
         return Observable.just(x + "-"); 
        } catch (Exception ex) { 
         return Observable.error(ex); 
        } 
       }).map(s -> { 
        if (s.equals("3-")) { 
         throw new IllegalArgumentException(); 
        } 
        return s + s; 
       }).take(1) 
         .zipWith(Observable.just("X"), (s, s2) -> s + s2) 
         .onErrorResumeNext(Observable.empty()); 
      }); 

    TestObserver<String> test = stringObservable.test(); 

    test.assertResult("1-1-X"); 
}

來源

2016-11-02 13:34:45

非常感謝你！我有另一個問題：所以如果我有十個變壓器，他們每個人都可以拋出異常，所以我應該用這個代碼包裝他們每個？ – corvax

一如既往的答案是：這取決於你想要做什麼。如果你不想只將map操作符放到新的observable上，而是要將其他操作符放到onErrorResumeNext之前。如果其中一個操作員失敗，則該值（例如2）將失敗，而返回空的可觀察值。如果你想處理每個操作員的備份值，你可以將子流重新包裝成一個新的流和平面圖。請提供更多代碼。 –

如何在RxJava出錯後繼續流式傳輸項目？

回答

相關問題