RxJava -连续无限流

Question

我看到了另一个类似的问题，但它并没有解决我的问题：

        Observable<Integer> resumeSequence = Observable.just(-1);       
        Observable.just(1,0,2,5)
          .flatMap(x -> 
                      Observable.just(10/x)
                      //.onErrorReturn(error -> -1)
                      .onErrorResumeNext(resumeSequence)
                      .onExceptionResumeNext(resumeSequence)
              )
//        .onErrorReturn(error -> -1)
          .onErrorResumeNext(resumeSequence)
          .onExceptionResumeNext(resumeSequence)
          .subscribe(System.out::println, 
                      throwable -> System.out.println("IN ERROR CALLBACK" + throwable));

理想情况下，简历应该在flatMap内部。输出应该打印4个数字&流不应该转到错误回调。

Answer 1

Observable.just对于已经计算的值非常有用。

你的代码

 Observable.just(1,0,2,5)
      .flatMap(x -> 
                  Observable.just(10/x)
                  .onErrorResumeNext(resumeSequence)
                  .onExceptionResumeNext(resumeSequence)
          )

可以这样重写：

 Observable.just(1,0,2,5)
      .flatMap(x -> {
                  int result = 10 / x;
                  return Observable.just(result)
                  .onErrorResumeNext(resumeSequence)
                  .onExceptionResumeNext(resumeSequence);
          })

因此，它解释了您的“主”Observable处于错误状态，并且完全感谢onExceptionResumeNext运算符，因为嵌套的Observable永远不会被使用。

你想要实现的是不完成可观察到的，然后就会出现错误。要做到这一点，嵌套的Observable可以在主Observable未完成时立即完成。要做到这一点，您必须使用fromCallable操作符推迟10 / x的计算：

 Observable<Integer> resumeSequence = Observable.just(-1);       
 Observable.just(1,0,2,5)
      // 10 / x will be computed only when this nested observable will subscribe
      .flatMap(x -> Observable.fromCompletable(() -> 10/x)
                  .onErrorResumeNext(resumeSequence)
                  .onExceptionResumeNext(resumeSequence)
          )
      .onErrorResumeNext(resumeSequence)
      .onExceptionResumeNext(resumeSequence)
      .subscribe(System.out::println, 
                  throwable -> System.out.println("IN ERROR CALLBACK" + throwable));