RxJava 的 retryWhen 意外地一直重试

问题描述

任何人都可以......

1. 解释为什么以下代码会在调用 retryWhen 后无限期地重试？
2. 更正代码以在 Random.nextBoolean() 时停止重试 返回 true？

我认为 retryWhen 有 50% 的几率因为 Random.nextBoolean() 而被调用。但是，一旦被调用，就会进入无限循环。

import io.reactivex.rxjava3.core.Observable
import kotlin.random.Random

fun main() {
    Observable
        .just(Unit)
        .flatMap {
            if (Random.nextBoolean()) {
                Observable.just(Unit)
            } else {
                Observable.error(Throwable())
            }.retryWhen {
                it.flatMap {
                    println("retryWhen")
                    Observable.just(Unit)
                }
            }
        }
        .subscribe()
}

输出：

retryWhen
retryWhen
retryWhen
⋮
(continued)

解决方法

1. 您已将 retryWhen 应用于 if 语句任一分支的执行结果。

2. 将 retryWhen 运算符链接到由 flatMap 生成的流：

例如

Observable
  .just(Unit)
  .flatMap {
    if(Random.nextBoolean()) {
      Observable.just(Unit)
    } else {
      Observable.error(Throwable())
    }
  }
  .retryWhen {
      it.flatMap {
        println("retryWhen")
        Observable.just(Unit)
      }
  }
  .subscribe()

,

请看同样的question and answer

复制答案：

retryWhen 在 Observer 订阅它时调用提供的函数，因此您有一个主序列伴随着一个序列，该序列发出主序列失败的 Throwable。您应该将逻辑组合到您在此 Observable 中获得的 Function 上，因此最后，一个 Throwable 将在另一端产生一个值。

我认为您的期望是：

Observable
            .just(Unit)
            .flatMap {
                if (Random.nextBoolean()) {
                    Observable.just(Unit)
                } else {
                    Observable.error(Throwable())
                }.retryWhen {
                    it.takeWhile {
                        Random.nextBoolean()
                    }
                        .doOnNext {
                            println("retryWhen")
                        }
                }
            }
            .subscribe()