使用 RxJava2 从一系列异步 HTTP 请求中创建具有背压的 flowable答案

【问题标题】：Create a flowable with backpressure from a sequence of async HTTP requests using RxJava2使用 RxJava2 从一系列异步 HTTP 请求中创建具有背压的 flowable
【发布时间】：2018-02-03 06:54:27
【问题描述】：

我想创建一个Flowable 来流式传输从异步 REST API 端点接收到的结果，该端点发送每个块的结果（需要发送另一个请求以使用上一个响应的标头获取剩余的结果列表） .它不想使用背压策略，而是仅在必要时（在 Flowable 中请求时）等待发送后续 HTTP 请求。

函数Flowable.generate 非常接近我想要的，但它似乎只适用于同步调用。是否有类似这样的异步调用解决方案，或者您有什么建议吗？

谢谢

【问题讨论】：

我认为stackoverflow.com/questions/28047272/handle-paging-with-rxjava 有解决方案的指针

标签： rx-java rx-java2

【解决方案1】：

我不确定我是否理解正确，但我想你想一个接一个地执行一个请求，并且你需要前一个请求中的一些数据来执行下一个请求。

假设我们有以下Flowable 对象：

Flowable<String> first = Flowable.fromCallable(() -> {
  Thread.sleep(2000);
  return "I need to be executed first";
});

Flowable<String> second = Flowable.fromCallable(() -> "I need to be executed later");

首先Flowable 会比较慢，但是我们想在first 之后执行second。

我们可以使用concat(...) 操作符来做到这一点，如下所示：

Flowable.concat(first, second).subscribe(System.out::println);

根据 RxJava 文档，concat 操作符从两个或多个 Observable 发出发射，而不将它们交错，因此请求将以正确的顺序执行。

您可能还想根据第一个请求的条件或在第二个Flowable 中使用第一个Flowable 中的一些数据来执行第二个请求。

您可以使用flatMap(...) 运算符来实现，如下所示：

first.flatMap(
    s -> s.equals("I need to be executed first")
         ? second
         : Flowable.empty())
    .subscribe(System.out::println);

在本例中，第二个Flowable 只会在第一个与我们想要的完全一致时发出。如果不是，则会发出一个空的Flowable。如果您想在第二个Flowable 中使用来自第一个Flowable 的数据，您可以在flatMap(...) 运算符中创建一个新的自定义Flowable，而不是像这样返回second Flowable：

first.flatMap(
    s -> s.equals("I need to be executed first")
        ? Flowable.fromCallable(() -> "I'm using data from the first Flowable: ".concat(s))
        : Flowable.empty())
    .subscribe(System.out::println);

希望这个回答对你有帮助。

【讨论】：