设为首页收藏本站
开启辅助访问

在路上

 找回密码
 立即注册
在路上 › 站点首页 ›学习 › 查看内容

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

2017-3-7 12:53| 发布者: zhangjf| 查看: 1347| 评论: 0

摘要: 原文链接： RxJava's repeatWhen and retryWhen, explained 原文作者： Daniel Lew 译文出自： 小?子的简书 译者： 小?子 状态： 完成 译者注：为了方便因Lambda（译文）还不够了解的同学进行阅 ...

原文链接： RxJava's repeatWhen and retryWhen, explained

原文作者： Daniel Lew

译文出自： 小?子的简书

译者： 小?子

状态： 完成

译者注：为了方便因Lambda（译文）还不够了解的同学进行阅读，本篇译文替换了原文中全部Lambda表达式。

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

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

Repeat与Retry的对比

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

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

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

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

Notification Handler

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

  1. retryWhen(Func1<? super Observable<? extends java.lang.Throwable>,? extends Observable<?>> notificationHandler)
复制代码

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

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

Func1像个工厂类，用来实现你自己的重试逻辑。

输入的是一个Observable。

输出的是一个Observable。

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

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

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

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

  1. source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
  2. @Override public Observable<?> call(Observable<? extends Throwable> errors) {
  4. return errors.flatMap(new Func1<Throwable, Observable<?>>() {
  5. @Override public Observable<?> call(Throwable error) {
  7. // For IOExceptions, we retry
  8. if (error instanceof IOException) {
  9. return Observable.just(null);
  10. }
  12. // For anything else, don't retry
  13. return Observable.error(error);
  14. }
  15. });
  16. }
  17. })
复制代码

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

经验之谈

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

.repeatWhen()与.retryWhen()非常相似，只不过不再响应onCompleted作为重试条件，而是onError。因为onCompleted没有类型，所有输入变为Observable。

每一次事件流的订阅notificationHandler（也就是Func1）只会调用一次。这也是讲得通的，因为你有一个可观测的Observable，它能够发送任意数量的error。

输入的Observable必须作为输出Observable的源。你必须对Observable做出反应，然后基于它发送事件；你不能只返回一个通用泛型流。

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

  1. .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
  2. @Override public Observable<?> call(Observable<? extends Throwable> errors) {
  4. return Observable.just(null);}
  5. })
复制代码

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

  1. .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
  2. @Override public Observable<?> call(Observable<? extends Throwable> errors) {
  4. return errors;
  5. }
  6. })
复制代码

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

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

使用方式

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

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

  1. source.repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
  2. @Override public Observable<?> call(Observable<? extends Void> completed) {
  4. return completed.delay(5, TimeUnit.SECONDS);
  5. }
  6. })
复制代码

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

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

  1. source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
  2. @Override public Observable<?> call(Observable<? extends Throwable> errors) {
  4. return errors.flatMap(new Func1<Throwable, Observable<?>>() {
  5. @Override public Observable<?> call(Throwable error) {
  7. return Observable.timer(5, TimeUnit.SECONDS);
  8. }
  9. });
  10. }
  11. })
复制代码

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

使用.flatMap() + .timer()实现有限次数的重订阅

  1. source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
  2. @Override public Observable<?> call(Observable<? extends Throwable> errors) {
  4. return errors.zipWith(Observable.range(1, 3), new Func2<Throwable, Integer, Integer>() {
  5. @Override public Integer call(Throwable throwable, Integer i) {
  7. return i;
  8. }
  9. });
  10. }
  11. })
复制代码

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

  1. zip(error1, 1) -> onNext(1) <-- Resubscribe
  2. zip(error2, 2) -> onNext(2) <-- Resubscribe
  3. zip(error3, 3) -> onNext(3) <-- Resubscribe
  4. onCompleted() <-- No resubscription
复制代码

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

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

  1. source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
  2. @Override public Observable<?> call(Observable<? extends Throwable> errors) {
  4. return errors.zipWith(Observable.range(1, 3), new Func2<Throwable, Integer, Integer>() {
  5. @Override public Integer call(Throwable throwable, Integer i) {
  7. return i;
  8. }
  9. }).flatMap(new Func1<Integer, Observable<? extends Long>>() {
  10. @Override public Observable<? extends Long> call(Integer retryCount) {
  12. return Observable.timer((long) Math.pow(5, retryCount), TimeUnit.SECONDS);
  13. }
  14. });
  15. }
  16. })
复制代码

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

收藏 邀请
上一篇：JVM工作原理下一篇：Grouping策略，并行度，消息的可靠处理机制

最新评论

相关分类

小黑屋|在路上 ( 蜀ICP备15035742号-1 )  

;

GMT+8, 2025-5-3 13:58

Copyright 2015-2025 djqfx

返回顶部