使用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請求序列創建一個流動背壓

回答

相關問題