RxJava - 继续无限流答案

【问题标题】：RxJava - Continue infinite streamRxJava - 继续无限流
【发布时间】：2016-12-08 10:55:59
【问题描述】：

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

        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 个数字，流程不应转到错误回调。

【问题讨论】：

你能改写你的问题吗，很清楚是什么问题？

标签： rx-java

【解决方案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 可以在您的主要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));

【讨论】：