RxJava不重試與觀察

Question

工作這是一個非常簡單的例子：

public static void main(String[] args) { 
    Observable.from(ListUtils.asList(1, 2, 3)).retry(3).subscribe(new Observer<Integer>() { 
     @Override 
     public void onCompleted() { 
      System.out.println("onCompleted"); 
     } 

     @Override 
     public void onError(Throwable e) { 
      System.out.println("onError"); 
     } 

     @Override 
     public void onNext(Integer integer) { 
      System.out.println(integer); 
      if (integer == 3) 
       throw new RuntimeException("onNext exception"); 

     } 
    }); 
} 

控制檯輸出爲：1,2,3，onerror的。 但我的預期：1,2,3,1,2,3,1,2,3，onError。

Answer 1

不是錯誤一旦發生從可觀察的用戶退訂，並且如果你使用操作符重試將重試僅如果操作員沒有在主要管道，但在flatMap操作

這一個，因爲使用重試是flatMap後，將工作

@Test 
public void retryInFlatMap() { 
    Observable.from(Arrays.asList(1, 2, 3, 4)) 
      .flatMap(number -> Observable.just(number) 
        .doOnNext(n -> { 
         if (n == 2) { 
          throw new NullPointerException(); 
         } 
        })) 
        .retry(3) 
      .subscribe(n-> System.out.println("number:"+n)); 
} 

這一個，因爲是一個地圖後，它不會

int cont=0; 
@Test 
public void retryInMap() { 
    Observable.from(Arrays.asList(1, 2, 3, 4)) 
      .map(number ->{ 
         if (cont == 2) { 
          throw new NullPointerException(); 
         } 
         cont++; 
         return number; 
        }) 
      .retry(3) 
      .subscribe(n-> System.out.println("number:"+n)); 
} 

如果你想看到更多EXA mples看看這裏https://github.com/politrons/reactive

Answer 2

retry()運營商重新訂閱並執行上面的過程時，observable發出錯誤。 請參閱this。

我寫了示例代碼。

Observable.from(Arrays.asList(1, 2, 3, 4)) 
    .flatMap(new Func1<Integer, Observable<Integer>>() { 
     @Override public Observable<Integer> call(Integer integer) { 
      if(integer == 4) { 
       return Observable.error(new Exception()); 
      } 
      return Observable.just(integer); 
     } 
    }) 
    // When error occurred, re-subscribe and execute above process 
    // If error occurred over 2 times, Observer.onError occurred 
    .retry(2) 
    .subscribe(new Observer<Integer>() { 
     @Override public void onCompleted() { 
      System.out.println("onCompleted"); 
     } 

     @Override public void onError(Throwable e) { 
      System.out.println("onError"); 
     } 

     @Override public void onNext(Integer integer) { 
      System.out.println(integer); 
     } 
    });