如何避免 Rx 流中重复的网络调用？答案

【问题标题】：How to avoid duplicate network calls in an Rx stream?如何避免 Rx 流中重复的网络调用？
【发布时间】：2017-10-03 05:39:58
【问题描述】：

我有以下流。

Observable.just(Unit) // execute immediately
    .mergeWith(tryAgainRelay) // execute again when this relay emits
    .flatMap {
      api.call() // emits sealed class single containing `Success` or `Error` state
          .toObservable()
          .startWith(Loading) // emit loading (same sealed class as above)
    }

但是，我从不希望重复（同时）网络调用（内部平面图）。如果 tryAgainRelay 在网络调用已经在运行时发出，我希望丢弃这些发射。我可以使用以下代码避免它们：

var isLoading = false // track whether api call is in flight
Observable.just(Unit)
    .mergeWith(tryAgainRelay)
    .filter { isLoading.not() } // prevent emissions if api call is already in flight
    .flatMap {
      api.call()
          .toObservable()
          .startWith(Loading)
          // update state variable
          .doOnNext {
            isLoading = when (it) {
              Loading -> true
              is Error, is Success -> false
            }
          }
        }

这有点难看，并且涉及使用.doOnNext() 走出流。有没有更好、更惯用的方法来做到这一点？

【问题讨论】：

你能在 api 调用中使用retry 代替吗？
重试有什么帮助？那只会影响 onError 对吧？
“如果 tryAgainRelay 在网络调用已在运行时发出，我希望丢弃此发出”在我看来，您应该尝试使用 skipUntil 运算符。但您可能不得不分享api.call().toObservable() 的排放量
使用switchMap() 运算符代替flatMap()。
@RafaelGuillen 没有。 switchMap 将有一个新的网络请求替换一个正在进行的请求。我希望在请求正在进行时丢弃新请求

标签： rx-java system.reactive rx-java2

【解决方案1】：

使用concatMap 而不是flatMap。

重试将一个接一个地按顺序“排队”创建，因此您可以确保只有一个请求将并行执行（下一个请求将在前一个请求完成时开始 - 完成或错误 - 和如果tryAgainRelay已经发出）。

【讨论】：