flatmap

RxJava 2.x中共有三個無縫相似的運算符： flatMap() ， concatMap()和concatMapEager() 。 它們都接受相同的參數-從原始流的單個項目到任意類型的（子）流的函數。 換句話說，如果您有Flowable<T>則可以為任意R類型提供從T到Flowable<R>的函數。 應用任何這些運算符后，您最終得到Flowable<R> 。 那么它們有何不同？

樣例項目

首先，讓我們構建一個示例應用程序。 我們將使用Retrofit2 HTTP客戶端包裝，該包裝具有RxJava2的內置插件。 我們的任務是利用GeoNames API來查找世界上任何城市的人口。 該界面如下所示：

public interface GeoNames {Flowable<Long> populationOf(String city);}

該接口的實現由Retrofit自動生成，向下滾動以查看膠粘源代碼。 暫時僅假設我們有一個函數，該函數采用具有城市名稱的String并異步返回具有該城市人口的單元素流。 還要假設我們有固定的城市要查找：

Flowable<String> cities = Flowable.just("Warsaw", "Paris", "London", "Madrid" );

我們的目標是獲取每個城市的人口。

帶有concatMap()的示例應用程序如下所示：

cities.concatMap(geoNames::populationOf).subscribe(response -> log.info("Population: {}", response));

在我們看到結果之前，讓我們研究一下concatMap()在做什么。 對于每個上游事件（ 城市 ），它都調用一個函數，該函數用（子）流替換該事件。 在我們的情況下，它是Long的一元流（ Flowable<Long> ）。 因此，與所有運算符進行比較之后，我們最終得到的是Long流（ Flowable<Flowable<Long>> ）流。 當我們分析操作員為展平此類嵌套流所做的操作時，就會出現真正的區別。

concatMap()將首先訂閱第一concatMap()流（ Flowable<Long>代表華沙的人口）。 訂閱實際上是指進行物理HTTP調用。 僅當第一concatMap()流完成時（在我們的情況下發出單個Long并發出完成信號）， concatMap()才會繼續。 繼續意味著訂閱第二個子流并等待其完成。 最后一個子流完成時，結果流完成。 這導致了以下信息流：1702139，2138551，7556900和3255944。因此，恰好是華沙，巴黎，倫敦和馬德里的人口。 輸出順序完全可以預測。 但是，它也是完全順序的。 完全沒有并發發生，只有在第一個HTTP結束時才進行第二個HTTP調用。 RxJava所增加的復雜性根本沒有回報：

23:33:33.531 | Rx-1 | --> GET .../searchJSON?q=Warsaw http/1.1 23:33:33.656 | Rx-1 | <-- 200 OK .../searchJSON?q=Warsaw (123ms) 23:33:33.674 | Rx-1 | Population: 1702139 23:33:33.676 | Rx-1 | --> GET .../searchJSON?q=Paris http/1.1 23:33:33.715 | Rx-1 | <-- 200 OK .../searchJSON?q=Paris (38ms) 23:33:33.715 | Rx-1 | Population: 2138551 23:33:33.716 | Rx-1 | --> GET .../searchJSON?q=London http/1.1 23:33:33.754 | Rx-1 | <-- 200 OK .../searchJSON?q=London (37ms) 23:33:33.754 | Rx-1 | Population: 7556900 23:33:33.755 | Rx-1 | --> GET .../searchJSON?q=Madrid http/1.1 23:33:33.795 | Rx-1 | <-- 200 OK .../searchJSON?q=Madrid (40ms) 23:33:33.796 | Rx-1 | Population: 3255944

如您所見，沒有多線程發生，請求是順序的，彼此等待。 從技術上講，并非所有這些都必須在同一線程中發生，但是它們絕不會重疊并且可以利用并發性。 最大的好處是可以保證結果事件的順序，一旦我們進入flatMap() ，這種順序就不那么明顯了……

flatMap()代碼幾乎完全相同：

cities.flatMap(geoNames::populationOf).subscribe(response -> log.info("Population: {}", response));

就像之前一樣，我們從Long流開始（ Flowable<Flowable<Long>> ）。 但是， flatMap()運算符渴望一次訂閱所有子流，而不是一個個地訂閱每個子流。 這意味著我們看到在不同線程中同時啟動多個HTTP請求：

00:10:04.919 | Rx-2 | --> GET .../searchJSON?q=Paris http/1.1 00:10:04.919 | Rx-1 | --> GET .../searchJSON?q=Warsaw http/1.1 00:10:04.919 | Rx-3 | --> GET .../searchJSON?q=London http/1.1 00:10:04.919 | Rx-4 | --> GET .../searchJSON?q=Madrid http/1.1 00:10:05.449 | Rx-3 | <-- 200 OK .../searchJSON (529ms) 00:10:05.462 | Rx-3 | Population: 7556900 00:10:05.477 | Rx-1 | <-- 200 OK .../searchJSON (557ms) 00:10:05.478 | Rx-1 | Population: 1702139 00:10:05.751 | Rx-4 | <-- 200 OK .../searchJSON (831ms) 00:10:05.752 | Rx-4 | Population: 3255944 00:10:05.841 | Rx-2 | <-- 200 OK .../searchJSON (922ms) 00:10:05.843 | Rx-2 | Population: 2138551

當任何基礎子流中的任何一個發出任何值時，它將立即向下游傳遞給訂戶。 這意味著我們現在可以在事件發生時即時處理事件。 請注意，結果流是亂序的。 我們收到的第一個事件是7556900，恰好是倫敦的人口，在第一流中排名第二。 與concatMap()相反， flatMap()無法保留順序，因此以“隨機”順序發出值。 好吧，不是真正隨機的，我們只是在值可用時立即接收它們。 在此特定執行中，首先是針對倫敦的HTTP響應，但絕對不能保證。 這導致一個有趣的問題。 我們有各種各樣的人口價值流和最初的城市流。 但是，輸出流可以是事件的任意排列，而且我們不知道哪個人口對應哪個城市。 我們將在后續文章中解決此問題。

concatMapEager()似乎兩全其美：并發性和輸出事件的有保證順序：

cities.concatMapEager(geoNames::populationOf).subscribe(response -> log.info("Population: {}", response));

在了解了concatMap()和flatMap()功能之后，了解concatMapEager()相當簡單。 急切地讓流concatMapEager()流（ duh！ ）同時預訂所有子流。 但是，此運算符可確保首先傳播第一個子流的結果，即使它不是要完成的第一個子流也是如此。 一個示例將Swift揭示這意味著什么：

00:34:18.371 | Rx-2 | --> GET .../searchJSON?q=Paris http/1.1 00:34:18.371 | Rx-3 | --> GET .../searchJSON?q=London http/1.1 00:34:18.371 | Rx-4 | --> GET .../searchJSON?q=Madrid http/1.1 00:34:18.371 | Rx-1 | --> GET .../searchJSON?q=Warsaw http/1.1 00:34:18.517 | Rx-3 | <-- 200 OK .../searchJSON?q=London (143ms) 00:34:18.563 | Rx-1 | <-- 200 OK .../searchJSON?q=Warsaw (189ms) 00:34:18.565 | Rx-1 | Population: 1702139 00:34:20.460 | Rx-2 | <-- 200 OK .../searchJSON?q=Paris (2086ms) 00:34:20.460 | Rx-4 | <-- 200 OK .../searchJSON?q=Madrid (2086ms) 00:34:20.461 | Rx-2 | Population: 2138551 00:34:20.462 | Rx-2 | Population: 7556900 00:34:20.462 | Rx-2 | Population: 3255944

我們立即啟動四個HTTP請求。 從日志輸出中，我們可以清楚地看到倫敦的居民首先被返回。 但是，訂戶沒有收到它，因為華沙尚未到來。 巧合的是，華沙排名第二，因此華沙人口可以在下游傳遞給訂戶。 不幸的是，倫敦人口必須等待更多，因為首先我們需要巴黎人口。 巴黎（緊隨其后是馬德里）完成后，所有剩余結果都將傳遞到下游。

請注意，即使人口充足，倫敦的人口也必須等待Hibernate，直到華沙和巴黎完成。 那么concatMapEager()是最好的并發運算符嗎？ 不完全的。 想象一下，我們有一個數千個城市的列表，每一個城市我們都獲取一張1MB的圖片。 使用concatMap()我們可以依次（即緩慢concatMap()下載圖片。 使用flatMap()可以同時下載圖片，并在圖片到達時盡快進行處理。 現在， concatMapEager()呢？ 在最壞的情況下，我們可以使用concatMapEager()緩存999張圖片，因為來自第一個城市的圖片恰好是最慢的。 即使我們已經擁有99.9％的結果，但由于我們執行嚴格的排序，因此我們無法對其進行處理。

使用哪個運算符？

flatMap()應該是您的首選武器。 它允許與流行為進行有效的并發。 但是要準備好接收亂序的結果。 僅當提供的轉換速度如此之快，順序處理不是問題時， concatMap()才能很好地工作。 concatMapEager()非常方便，但是要注意內存消耗。 同樣在最壞的情況下，您可能最終會閑置，等待很少的響應。

附錄：配置Retrofit2客戶端

實際上，我們在本文中始終使用的GeoNames服務接口如下所示：

public interface GeoNames {@GET("/searchJSON")Single<SearchResult> search(@Query("q") String query,@Query("maxRows") int maxRows,@Query("style") String style,@Query("username") String username);default Flowable<Long> populationOf(String city) {return search(city, 1, "LONG", "s3cret").map(SearchResult::getGeonames).map(g -> g.get(0)).map(Geoname::getPopulation).toFlowable();}}

非默認方法的實現由Retrofit2自動生成。 請注意，為簡單起見， populationOf()返回一個元素的Flowable<Long> 。 但是，要完全擁抱此API的本質，在現實世界中，其他實現將更為合理。 首先， SearchResult類返回結果的有序列表（省略getter / setter）：

class SearchResult {private List<Geoname> geonames = new ArrayList<>(); }class Geoname {private double lat;private double lng;private Integer geonameId;private Long population;private String countryCode;private String name; }

畢竟，世界上有許多華沙和倫敦 。 我們默默地假設列表將包含至少一個元素，而第一個是正確的匹配。 更合適的實現應返回所有匹配，甚至返回更好的Maybe<Long>類型以反映沒有匹配項：

default Maybe<Long> populationOf(String city) {return search(city, 1, "LONG", "nurkiewicz").flattenAsFlowable(SearchResult::getGeonames).map(Geoname::getPopulation).firstElement(); }

粘合代碼如下所示。 首先Jackson的設置，以便解析來自API的響應：

import com.fasterxml.jackson.databind.ObjectMapper;private ObjectMapper objectMapper() {return new ObjectMapper().configure(FAIL_ON_UNKNOWN_PROPERTIES, false); }

FAIL_ON_UNKNOWN_PROPERTIES通常是您想要的。 否則，您必須映射JSON響應中的所有字段，并且當API生產者引入新的或向后兼容的字段時，代碼將中斷。 然后我們設置OkHttpClient ，由Retrofit在下面使用：

import okhttp3.OkHttpClient; import okhttp3.logging.HttpLoggingInterceptor;private OkHttpClient client() {HttpLoggingInterceptor interceptor = new HttpLoggingInterceptor();interceptor.setLevel(HttpLoggingInterceptor.Level.BASIC);return new OkHttpClient.Builder().addInterceptor(interceptor).build(); }

有時您可以跳過OkHttp客戶端的配置，但是我們添加了日志記錄攔截器。 默認情況下，OkHttp使用java.util.logging日志記錄，因此為了使用體面的日志記錄框架，我們必須在開始時就安裝橋：

import org.slf4j.bridge.SLF4JBridgeHandler;static {SLF4JBridgeHandler.removeHandlersForRootLogger();SLF4JBridgeHandler.install(); }

最后進行自我改造：

import io.reactivex.schedulers.Schedulers; import retrofit2.Retrofit; import retrofit2.adapter.rxjava2.RxJava2CallAdapterFactory; import retrofit2.converter.jackson.JacksonConverterFactory;GeoNames createClient() {return new Retrofit.Builder().client(client()).baseUrl("http://api.geonames.org").addCallAdapterFactory(RxJava2CallAdapterFactory.createWithScheduler(Schedulers.io())).addConverterFactory(JacksonConverterFactory.create(objectMapper())).build().create(GeoNames.class); }

調用createClient()將產生GeoNames接口的動態實現。 我們使用了以下依賴項：

compile 'io.reactivex.rxjava2:rxjava:2.0.6'compile 'com.squareup.retrofit2:adapter-rxjava2:2.3.0' compile 'com.squareup.retrofit2:converter-jackson:2.0.1' compile 'com.squareup.okhttp3:logging-interceptor:3.8.0'compile 'ch.qos.logback:logback-classic:1.1.7' compile 'org.slf4j:slf4j-api:1.7.21' compile 'org.slf4j:jul-to-slf4j:1.7.21'

翻譯自: https://www.javacodegeeks.com/2017/08/flatmap-vs-concatmap-vs-concatmapeager-rxjava-faq.html

flatmap

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的flatmap_flatMap（）与concatMap（）与concatMapEager（）– RxJava常见问题解答的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。