使用RxJava2从异步HTTP请求序列创建一个流动背压

我想创建一个Flowable来传输从异步REST API端点接收到的结果，该端点发送每个块的结果（必须发送另一个请求以获取使用前一个响应中的头部的其余结果列表）。它不想使用背压策略，而是只在必要时才发送后续HTTP请求（在Flowable中请求时）。使用RxJava2从异步HTTP请求序列创建一个流动背压

函数Flowable.generate非常接近我想要的，但它似乎只适用于同步调用。有没有类似于这个异步调用的解决方案，或者你有任何建议来做到这一点？

感谢

来源

2017-08-25 Jean-Baptiste Reich

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

我不知道如果我理解正确的，你的，但我想你要陆续执行一个请求，你需要从之前的请求的一些数据来执行一个请求。

比方说，我们有以下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文档，毗连运算发出来自两个或多个观测量的排放没有交错他们，所以请求将按照正确的顺序执行。

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

可以实现与flatMap(...)操作如下：

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

在这个例子中，第二Flowable只会发出，当第一个是完全一样的，我们要。如果不是，则会发射一个空的Flowable。如果你想在第二个从第一Flowable使用数据，您可以创建flatMap(...)运营商内部的一个新的自定义Flowable而不是只返回secondFlowable这样的：

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);

我希望，这个答案将有助于您。

来源

2017-08-27 10:56:51

使用RxJava2从异步HTTP请求序列创建一个流动背压

回答

相关问题