对RxJava中.repeatWhen()和.retryWhen()操作符的思考

﹏ヽ暗。殇╰゛Y 2022-06-03 00:16 35阅读 0赞

>  *  原文链接： [RxJava's repeatWhen and retryWhen, explained][RxJava_s repeatWhen and retryWhen_ explained]
>  *  原文作者： [Daniel Lew][]
>  *  译文出自： [小鄧子的简书][Link 1]
>  *  译者： 小鄧子
>  *  状态： 完成
>  *  译者注：为了方便因[Lambda][]（[译文][Link 2]）还不够了解的同学进行阅读，本篇译文替换了原文中全部Lambda表达式。

第一次见到[.repeatWhen()][.repeatWhen]和[.retryWhen()][.retryWhen]这两个操作符的时候就非常困惑了。不得不说，它们绝对是“最令人困惑弹珠图”的有力角逐者。

![500][]

然而它们都是非常有用的操作符：允许你有条件的重新订阅已经结束的`Observable`。我最近研究了它们的工作原理，现在我希望尝试着去解释它们（因为，我也是耗费了一些精力才参透它们）。

## Repeat与Retry的对比 ##

首先，来了解一下[.repeat()][.repeat]和[.retry()][.retry]之间最直观的区别是什么？这个问题并不难：区别就在于什么样的终止事件会触发重订阅。

当.repeat()接收到.onCompleted()事件后触发重订阅。

当.retry()接收到.onError()事件后触发重订阅。

然而，这种简单的叙述尚不能令人满意。试想如果你要实现一个延迟数秒的重订阅该如何去做？或者想通过观察错误来决定是否应该重订阅呢？这种情况下就需要`.repeatWhen()`和`.retryWhen()`的介入了，因为它们允许你为重试提供自定义逻辑。

## Notification Handler ##

你可以通过一个叫做`notificationHandler`的函数来实现重试逻辑。这是`.retryWhen()`的方法签名（译者注：方法签名，指方法名称、参数类型和参数数量等）：

retryWhen(Func1<? super Observable<? extends java.lang.Throwable>,? extends Observable<?>> notificationHandler)

签名很长，甚至不能一口气读完。我发现它很难理解的原因是因为存在一大堆的泛型约定。

简化后，它包括三个部分：

1.  `Func1`像个工厂类，用来实现你自己的重试逻辑。
2.  输入的是一个`Observable<Throwable>`。
3.  输出的是一个`Observable<?>`。

首先，让我们来看一下最后一部分。被返回的`Observable<?>`所要发送的事件决定了重订阅是否会发生。如果发送的是`onCompleted`或者`onError`事件，将不会触发重订阅。相对的，如果它发送`onNext`事件，则触发重订阅（不管`onNext`实际上是什么事件）。这就是为什么使用了通配符作为泛型类型：这仅仅是个通知（next, error或者completed），一个很重要的通知而已。

source每次一调用`onError(Throwable)`，`Observable<Throwable>`都会被作为输入传入方法中。换句话说就是，它的每一次调用你都需要决定是否需要重订阅。

当订阅发生的时候，工厂`Func1`被调用，从而准备重试逻辑。那样的话，当`onError`被调用后，你已经定义的重试逻辑就能够处理它了。

这里有个例子展示了我们应该在哪些场景下订阅source，比如，只有在`Throwable`是`IOException`的情况下请求重订阅，否则不（重订阅）。

source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
              @Override public Observable<?> call(Observable<? extends Throwable> errors) {
    
                return errors.flatMap(new Func1<Throwable, Observable<?>>() {
                  @Override public Observable<?> call(Throwable error) {
    
                    // For IOExceptions, we  retry
                    if (error instanceof IOException) {
                      return Observable.just(null);
                    }
    
                    // For anything else, don't retry
                    return Observable.error(error);
                  }
                });
              }  
            })

由于每一个error都被flatmap过，因此我们不能通过直接调用`.onNext(null)`触发重订阅或者`.onError(error)`来避免重订阅。

## 经验之谈 ##

这里有一些关于`.repeatWhen()`和`.retryWhen()`的要点，我们应该牢记于心。

*  `.repeatWhen()`与`.retryWhen()`非常相似，只不过不再响应`onError`作为重试条件，而是`onCompleted`。因为`onCompleted`没有类型，所有输入变为`Observable<Void>`。
 *  每一次事件流的订阅`notificationHandler`（也就是`Func1`）只会调用一次。这也是讲得通的，因为你有一个可观测的`Observable<Throwable>`，它能够发送任意数量的error。
 *  输入的`Observable`必须作为输出`Observable`的源。你必须对`Observable<Throwable>`做出反应，然后基于它发送事件；你不能只返回一个通用泛型流。

换言之就是，你不能做类似的操作：

.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                  @Override public Observable<?> call(Observable<? extends Throwable> errors) {
    
                    return Observable.just(null);}
                })

因为它不仅不能奏效，而且还会打断你的链式结构。你应该做的是，而且至少应该做的是，把输入作为结果返回，就像这样：

.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                  @Override public Observable<?> call(Observable<? extends Throwable> errors) {
    
                    return errors;
                  }
                })

（顺便提一下，这在逻辑上与单纯使用`.retry()`操作符的效果是一样哒）

*  输入`Observable`只在终止事件发生的时候才会触发（对于`.repeatWhen()`来说是`onCompleted`,而对于`.retryWhen()`来说是`onError`）。它不会从源中接收到任何`onNext`的通知，所以你不能通过观察被发送的事件来决定重订阅。如果你真的需要这样做，你应该添加像`.takeUntil()`这样的操作符，来拦截事件流。

## 使用方式 ##

现在，假设你已大概了解了`.repeatWhen()`和`.retryWhen()`，那么你能将一些什么样的精简逻辑放入到`notificationHandler`中呢？

使用`.repeatWhen()` \+ `.delay()`定期轮询数据:

source.repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
                  @Override public Observable<?> call(Observable<? extends Void> completed) {
    
                    return completed.delay(5, TimeUnit.SECONDS);
                  }
                })

直到`notificationHandler`发送`onNext()`才会重订阅到source。因为在发送`onNext()`之前`delay`了一段时间，所以优雅的实现了延迟重订阅，从而避免了不间断的数据轮询。

非此即彼，使用`.flatMap()` \+ `.timer()`实现延迟重订阅:  
(译者注：在RxJava 1.0.0及其之后的版本，官方已不再提倡使用`.timer()`操作符，因为`.interval()`具有同样的功能)

source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                  @Override public Observable<?> call(Observable<? extends Throwable> errors) {
    
                    return errors.flatMap(new Func1<Throwable, Observable<?>>() {
                      @Override public Observable<?> call(Throwable error) {
    
                        return Observable.timer(5, TimeUnit.SECONDS);
                      }
                    });
                  }
                })

当需要与其他逻辑协同的时候，这种替代方案就变得非常有用了，比如。。。

使用`.zip()` \+ `.range()`实现有限次数的重订阅

最后的结果就是每个error都与`range`中一个输出配对出现，就像这样：

zip(error1, 1) -> onNext(1)  <-- Resubscribe  
    zip(error2, 2) -> onNext(2)  <-- Resubscribe  
    zip(error3, 3) -> onNext(3)  <-- Resubscribe  
    onCompleted()                <-- No resubscription

因为当第四次error出现的时候，`range(1,3)`中的数字已经耗尽了，所以它隐式调用了`onCompleted()`，从而导致整个`zip`的结束。防止了进一步的重试。

将可变延迟策略与次数限制的重试机制结合起来

source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                  @Override public Observable<?> call(Observable<? extends Throwable> errors) {
    
                    return errors.zipWith(Observable.range(1, 3), new Func2<Throwable, Integer, Integer>() {
                      @Override public Integer call(Throwable throwable, Integer i) {
    
                        return i;
                      }
                    }).flatMap(new Func1<Integer, Observable<? extends Long>>() {
                      @Override public Observable<? extends Long> call(Integer retryCount) {
                        
                        return Observable.timer((long) Math.pow(5, retryCount), TimeUnit.SECONDS);
                      }
                    });
                  }
                })

在这种用例的比较上，我认为`.flatMap()`\+`.timer()`的组合比单纯使用`.delay()`更可取，因为我们可以通过重试次数来修改延迟时间。重试三次，并且每一次的重试时间都是`5 ^ retryCount`，仅仅通过一些操作符的组合就帮助我们实现了指数退避算法（译者注：可参考[二进制指数退避算法][Link 3]）。

[RxJava_s repeatWhen and retryWhen_ explained]: https://link.jianshu.com/?t=http://blog.danlew.net/2016/01/25/rxjavas-repeatwhen-and-retrywhen-explained/
[Daniel Lew]: https://link.jianshu.com/?t=http://blog.danlew.net/
[Link 1]: https://www.jianshu.com/users/df40282480b4/latest_articles
[Lambda]: https://link.jianshu.com/?t=http://javarevisited.blogspot.sg/2014/02/10-example-of-lambda-expressions-in-java8.html
[Link 2]: https://link.jianshu.com/?t=http://www.importnew.com/16436.html
[.repeatWhen]: https://link.jianshu.com/?t=http://reactivex.io/RxJava/javadoc/rx/Observable.html#repeatWhen%28rx.functions.Func1%29
[.retryWhen]: https://link.jianshu.com/?t=http://reactivex.io/RxJava/javadoc/rx/Observable.html#retryWhen%28rx.functions.Func1%29
[500]: /images/20220603/80ff795f406d464ca4836e5845f7e783.png
[.repeat]: https://link.jianshu.com/?t=http://reactivex.io/RxJava/javadoc/rx/Observable.html#repeat%28%29
[.retry]: https://link.jianshu.com/?t=http://reactivex.io/RxJava/javadoc/rx/Observable.html#retry%28%29
[Link 3]: https://link.jianshu.com/?t=http://baike.baidu.com/view/2984826.htm