之前在做一個(gè)報(bào)表導(dǎo)出功能時(shí)我第一次真正被扁平化操作符難住。業(yè)務(wù)本身不復(fù)雜拿到一批訂單ID每個(gè)ID要異步去查詢訂單明細(xì)明細(xì)又是多條記錄最終把所有明細(xì)合并成一個(gè)數(shù)據(jù)流寫進(jìn)Excel。我一開始用 map 去寫結(jié)果拿回來的不是訂單明細(xì)流而是流里面套流FluxFlux 類型看著就難受。換成 flatMap 之后數(shù)據(jù)是對(duì)了但輸出順序全亂了再換 concatMap順序是保住了可速度直線下降。也就是從那天起我對(duì) flatMap 和 concatMap 的態(tài)度從會(huì)用變成了必須從底層搞懂。這篇就把 Reactor 3 里數(shù)據(jù)扁平處理的完整鏈路講清楚為什么響應(yīng)式編程里總繞不開 flatMap 和 concatMap它們底層到底怎么處理事件流什么時(shí)候用哪個(gè)以及我在真實(shí)項(xiàng)目里踩過的一些坑。適合已經(jīng)掌握 map、filter 等基礎(chǔ)操作符、但一碰到流里面套流就犯迷糊的讀者。1. 從 map 到 flatMap一次不得不做的思維轉(zhuǎn)換1.1 map 的邊界只能做 1:1 的同步變換很多剛接觸 Reactor 的同學(xué)會(huì)把 flatMap 想得很神秘其實(shí)它和 map 之間的關(guān)系非常天然。map 做的事情很單純上游來一個(gè)元素我同步地把它變成另一個(gè)元素一對(duì)一不改變流的維度。Flux.just(o-1, o-2, o-3) .map(orderId - orderId.toUpperCase()) .subscribe(System.out::println);這段代碼里每個(gè)字符串進(jìn)來出去還是字符串流的形狀沒有變。這種 1:1 變換在同步場(chǎng)景下非常舒服你不需要考慮異步、不需要考慮順序因?yàn)?map 本身就是阻塞的。但實(shí)際業(yè)務(wù)里一個(gè)元素映射出一個(gè)結(jié)果的情況遠(yuǎn)遠(yuǎn)不夠。你往往碰到的是一個(gè)元素對(duì)應(yīng)著一個(gè)異步操作而異步操作的結(jié)果又是一個(gè)容器。比如訂單ID要查詢出明細(xì)列表再比如要訪問某個(gè)服務(wù)去拿配置數(shù)據(jù)返回的是一個(gè)集合。這時(shí)候如果你繼續(xù)用 map寫出來的代碼會(huì)變成什么樣Flux 里的每個(gè)元素被映射成一個(gè) Publisher整個(gè)流就變成了嵌套流。Reactor 的泛型會(huì)把這種結(jié)構(gòu)完整地暴露出來編譯器會(huì)直接告訴你這個(gè)嵌套關(guān)系存在但你的下游訂閱者并不關(guān)心這件事它只想拿到扁平的訂單明細(xì)。這就是扁平化操作符存在的理由把每個(gè)上游元素 - 一個(gè)內(nèi)部流這種結(jié)構(gòu)重新壓平成一個(gè)對(duì)外統(tǒng)一發(fā)射元素的下游流。map 管不了這件事因?yàn)樗Y(jié)構(gòu)上就沒有展開這一層。1.2 當(dāng)每個(gè)元素自己開出一個(gè)流用一個(gè)最常見的場(chǎng)景來感受一下這種流里套流。假設(shè)上游是一串用戶ID下游是用戶詳情對(duì)象每個(gè)用戶ID都需要去遠(yuǎn)程接口拉一次數(shù)據(jù)。遠(yuǎn)程接口的函數(shù)簽名是這樣的FluxUserDetail fetchDetail(String userId);如果你堅(jiān)持用 mapFlux.just(u-1, u-2, u-3) .map(this::fetchDetail) // 返回 FluxFluxUserDetail靜態(tài)類型已經(jīng)清楚地告訴你map 只是把每個(gè)字符串換成了另一個(gè)流對(duì)象這些流對(duì)象彼此獨(dú)立并沒有合并到你想要的同一個(gè)下游流里。你把這樣的流交出去訂閱者要面對(duì)的是一層嵌套結(jié)構(gòu)處理起來極其別扭。用 flatMap 就不是這個(gè)效果了。同樣是這個(gè)映射函數(shù)flatMap 會(huì)自動(dòng)訂閱每個(gè)由用戶ID產(chǎn)生出來的內(nèi)部流并把它們發(fā)出的元素重新匯聚到一個(gè)外層的公共流中Flux.just(u-1, u-2, u-3) .flatMap(this::fetchDetail) .subscribe(detail - System.out.println(detail.getName()));從訂閱者的視角看它拿到的就是一個(gè)不斷的 UserDetail 流至于這個(gè)元素是 u-1 來的還是 u-3 來的它根本不關(guān)心。這也是扁平化這個(gè)詞最直觀的含義把二維的流結(jié)構(gòu)拉平成一維。打個(gè)比方map 就像一個(gè)人遞給你一張紙flatMap 則像他遞給你一疊紙同時(shí)幫你把紙攤開鋪到桌面上。鋪開之后你看不到原來的一疊這個(gè)層級(jí)只看到一張張獨(dú)立的紙張。1.3 扁平化操作符家族不止 flatMap 和 concatMapReactor 3 里和扁平化相關(guān)的操作符其實(shí)有一族除了標(biāo)題里的 flatMap、concatMap還有 flatMapSequential、switchMap 這些。它們的共性都是上游一個(gè)元素映射為一個(gè)內(nèi)部 Publisher區(qū)別在于合并內(nèi)部流的方式以及對(duì)順序和并發(fā)的取舍。我在實(shí)際項(xiàng)目里真正高頻使用的也就三四個(gè)flatMap 解決并發(fā)展開問題concatMap 解決順序強(qiáng)制問題flatMapSequential 在兩者之間找平衡switchMap 則適合最新結(jié)果覆蓋舊結(jié)果的場(chǎng)景。這篇先把 flatMap 和 concatMap 講透因?yàn)樗鼈兇砹吮馄交飪蓚€(gè)最核心的極端完全并發(fā)無序和完全串行有序。理解了這兩個(gè)極端flatMapSequential 和 switchMap 你自然而然就能推出來。2. flatMap 核心機(jī)制并發(fā)展開與無序合并2.1 一樣的代碼不一樣的輸出順序flatMap 最迷惑人的地方在于它表面上看起來和 concatMap 的寫法幾乎一樣就是方法名差一個(gè)前綴但行為差異巨大。先用一個(gè)簡(jiǎn)單的延遲模擬來觀察Flux.just(u-1, u-2, u-3) .flatMap(this::fetchName) .subscribe(System.out::println);對(duì)應(yīng)的映射函數(shù)給不同 ID 制造不同延遲private MonoString fetchName(String id) { return Mono.delay(Duration.ofMillis(u-1.equals(id) ? 80 : 20)) .map(tick - id -detail); }我先賣個(gè)關(guān)子你猜一下輸出順序是什么。直覺可能會(huì)告訴你既然上游發(fā)射順序是 u-1、u-2、u-3那輸出應(yīng)該也是這個(gè)順序。但 flatMap 的答案恰恰反直覺u-1 雖然先進(jìn)入流但它延遲了 80msu-2 和 u-3 的延遲只有 20ms所以先輸出的反而是后面的元素。這就是 flatMap 的核心特征它會(huì)立刻訂閱所有內(nèi)部流多個(gè)內(nèi)部流之間的發(fā)射順序是交錯(cuò)的誰先有結(jié)果誰先往公共流里發(fā)。如果你依賴先查詢的先輸出這種假設(shè)用 flatMap 就會(huì)踩到順序坑。2.2 flatMap 內(nèi)部做了什么多路訂閱與合并從源碼層面看flatMap 做的事情可以簡(jiǎn)化成三步。第一步上游每到達(dá)一個(gè)元素就通過映射函數(shù)生成一個(gè)對(duì)應(yīng)的內(nèi)部 Publisher第二步框架立即訂閱這個(gè)內(nèi)部 Publisher并且這個(gè)訂閱是并發(fā)的不是等上一個(gè)內(nèi)部流結(jié)束再訂閱下一個(gè)第三步所有并發(fā)運(yùn)行中的內(nèi)部流誰發(fā)射了元素誰就把元素丟進(jìn)同一個(gè)合并緩沖區(qū)再傳給下游。這個(gè)合并緩沖區(qū)就是無序的根源。它本質(zhì)上是一種 merge 行為多個(gè)生產(chǎn)者在同時(shí)生產(chǎn)生產(chǎn)者之間沒有順序協(xié)議誰先生產(chǎn)完誰先繞過緩沖區(qū)抵達(dá)下游。所以在實(shí)際運(yùn)行中你可能看到u-2 的結(jié)果先出來u-3 的結(jié)果跟上u-1 的結(jié)果最后才到。這不是偶發(fā)而是 flatMap 的常態(tài)化表現(xiàn)。2.3 背壓、并發(fā)度與合并緩沖flatMap 還有一個(gè)容易被人忽略的重載參數(shù)并發(fā)度。默認(rèn)情況下flatMap 最多同時(shí)訂閱 256 個(gè)內(nèi)部流也就是說上游來元素的速度再快它也只會(huì)優(yōu)先處理 256 個(gè)其余的進(jìn)入排隊(duì)等待。flatMap(this::fetchDetail, 16); // 限制最多16個(gè)內(nèi)部流并發(fā)為什么要限制并發(fā)度我見過不少新人在并發(fā)量小的測(cè)試?yán)锱艿猛靡簧仙a(chǎn)就把下游數(shù)據(jù)庫連接池打爆。上游可能一瞬間來一萬個(gè)IDflatMap 默認(rèn)并發(fā) 256加上每個(gè) ID 的查詢要占用一條數(shù)據(jù)庫連接等于瞬時(shí)需要 256 個(gè)連接很多服務(wù)根本扛不住。這時(shí)候用第二參數(shù)把并發(fā)度調(diào)到 16、32既保留并發(fā)加速又保護(hù)下游資源。如果你還需要更精細(xì)的控制flatMap 還有第三個(gè)參數(shù) prefetch控制每個(gè)內(nèi)部流預(yù)取多少元素。日常業(yè)務(wù)里先調(diào)整 concurrency 就夠了prefetch 一般用默認(rèn)值即可。2.4 flatMap 最適用的場(chǎng)景和明顯的反例我個(gè)人的經(jīng)驗(yàn)flatMap 適合下面這幾類場(chǎng)景每個(gè)元素映射出的內(nèi)部流耗時(shí)不一致你希望更快的結(jié)果先出來不在乎順序。下游處理邏輯不依賴上游發(fā)射順序比如把數(shù)據(jù)寫入消息隊(duì)列消息之間互相獨(dú)立。你想利用并發(fā)加速把串行等待變成并行等待。反過來如果存在以下信號(hào)就別用 flatMap下游對(duì)處理順序有硬性要求比如必須按輸入ID順序?qū)慐xcel。內(nèi)部流之間互相依賴第二個(gè)元素的計(jì)算要等第一個(gè)元素的結(jié)果。前端需要展示先來先顯示的列表但帶上了明確的順序預(yù)期。順序要求這種東西剛開始寫響應(yīng)式代碼時(shí)容易被忽略因?yàn)樾?shù)據(jù)量下 flatMap 的亂序可能完全看不出來順序碰巧和輸入一致。一旦數(shù)據(jù)量上來或者某個(gè)服務(wù)偶發(fā)慢查詢亂序問題就會(huì)突然暴露而且這種 bug 非常難定位。3. concatMap 的順序保證從哪里來串行訂閱與它的成本3.1 concatMap 的慢是很誠實(shí)的concatMap 同樣是把上游元素映射成內(nèi)部流但它的合并方式完全不同。它不會(huì)并發(fā)訂閱所有內(nèi)部流而是嚴(yán)格地一個(gè)接一個(gè)先訂閱第一個(gè)元素生成的內(nèi)部流等這個(gè)內(nèi)部流完完整整地發(fā)出最后一個(gè)元素并結(jié)束再訂閱第二個(gè)元素生成的內(nèi)部流。所以你看到的輸出序列和上游輸入序列永遠(yuǎn)是嚴(yán)格對(duì)應(yīng)的。還是剛才那個(gè)延遲模擬Flux.just(u-1, u-2, u-3) .concatMap(this::fetchName) .subscribe(System.out::println);不管 u-1 內(nèi)部的查詢有多慢u-2 的結(jié)果都只能在 u-1 的流完全結(jié)束后才開始處理。這就是 wait-for-complete 的行為。它保證了順序代價(jià)是徹底放棄了并行。用生活類比來說flatMap 是超市開了多個(gè)收銀臺(tái)大家誰排完隊(duì)誰先走concatMap 是只有一個(gè)收銀臺(tái)第一個(gè)人結(jié)完賬走了第二個(gè)人才能開始把商品放上收銀臺(tái)。3.2 concatMap 的預(yù)取機(jī)制為了不斷流如果你讀過 concatMap 的實(shí)現(xiàn)會(huì)發(fā)現(xiàn)它沒有想象中那么死板它內(nèi)部其實(shí)有一個(gè)預(yù)取機(jī)制默認(rèn)預(yù)取數(shù)量是 2。也就是說concatMap 當(dāng)前正在處理第一個(gè)內(nèi)部流時(shí)上游可以提前把下一個(gè)元素準(zhǔn)備好這樣一旦第一個(gè)流結(jié)束第二個(gè)流立刻就能接著訂閱不用等上游重新發(fā)射數(shù)據(jù)。這個(gè)設(shè)計(jì)本質(zhì)上是為了解決串行場(chǎng)景下的空轉(zhuǎn)問題。它讓 concatMap 在保序的同時(shí)盡量保持流式節(jié)奏但千萬不要誤解成 concatMap 允許并行它只是做了流水線式的緩沖執(zhí)行仍然是串行。concatMap(this::fetchName, 4); // 增大預(yù)取稍微減小切換開銷調(diào)大預(yù)取量在內(nèi)部流很短、切換頻繁的場(chǎng)景下有點(diǎn)作用但作用有限核心瓶頸還是被串行執(zhí)行拖住的。3.3 做一個(gè)簡(jiǎn)單的延遲賬本為了直觀理解 concatMap 的成本我們算一筆賬。假設(shè)上游有 10 個(gè)元素每個(gè)元素對(duì)應(yīng)的內(nèi)部操作平均耗時(shí) 200ms。用 flatMap 并發(fā)執(zhí)行理論上總耗時(shí)接近最慢的那個(gè)元素大約 200 到 300ms 就能全部完成。用 concatMap 串行執(zhí)行總耗時(shí)是 10 乘以 200ms也就是約 2000ms。這筆延遲賬在元素多、單個(gè)操作慢的場(chǎng)景下會(huì)被放得很大。所以我會(huì)在代碼評(píng)審時(shí)特別留意如果有人把 flatMap 直接改成 concatMap 來修亂序 bug那他其實(shí)是在用性能換順序這種取舍不是不行但要想清楚代價(jià)。3.4 concatMap 真正不可替代的場(chǎng)景既然 concatMap 這么慢那什么時(shí)候必須用它我的經(jīng)驗(yàn)是凡是內(nèi)部流存在時(shí)間相關(guān)性、強(qiáng)依賴性的場(chǎng)景都逃不開它。比如我做過一個(gè)設(shè)備配置下發(fā)功能一批配置項(xiàng)之間是有先后約束的先寫網(wǎng)絡(luò)參數(shù)再重啟設(shè)備最后校驗(yàn)狀態(tài)。如果并發(fā)執(zhí)行第二步可能在第一步還沒完成時(shí)就啟動(dòng)整個(gè)流程都會(huì)出錯(cuò)。這種場(chǎng)景下concatMap 的順序保證不是性能問題而是正確性問題。再比如金融業(yè)務(wù)里的流水處理依賴上一筆狀態(tài)變更完成后才能處理下一筆的規(guī)則這種場(chǎng)景你不可能用 flatMap 去賭運(yùn)行時(shí)的順序。有一類更隱蔽的場(chǎng)景也適合 concatMap內(nèi)部流本身有副作用并且副作用之間存在上下文關(guān)聯(lián)。比如依次給用戶發(fā)提醒短信同一個(gè)用戶的兩條短信順序顛倒了后端不報(bào)錯(cuò)但前端客服會(huì)看到歷史順序混亂。這種邏輯層面的關(guān)聯(lián)靠測(cè)試不一定能覆蓋但它真實(shí)存在于業(yè)務(wù)語義里。4. 實(shí)戰(zhàn)選擇我在代碼里按這三條規(guī)則挑操作符4.1 規(guī)則一先回答順序重不重要這是我選型時(shí)問自己的第一句話。判斷標(biāo)準(zhǔn)很簡(jiǎn)單把輸入順序和輸出結(jié)果順序打亂下游能感知到變化嗎能感知就必須放棄 flatMap轉(zhuǎn)而考慮 concatMap 或 flatMapSequential感知不到就大膽用 flatMap 享受并發(fā)優(yōu)勢(shì)。有些場(chǎng)景的順序要求藏在細(xì)節(jié)里比如你要把處理結(jié)果回寫到一個(gè)數(shù)組數(shù)組索引按輸入順序排。這種情況下即使業(yè)務(wù)看起來不敏感寫入位置一變就會(huì)錯(cuò)位本質(zhì)還是順序敏感。下面這個(gè)表是我在評(píng)審時(shí)經(jīng)常用來和同事對(duì)齊的場(chǎng)景特征保序需求并發(fā)需求首選操作符數(shù)據(jù)上報(bào)、日志清洗無高flatMap依賴嚴(yán)格的流水處理高低concatMap結(jié)果展示、報(bào)表導(dǎo)出高高flatMapSequential秒級(jí)超時(shí)的實(shí)時(shí)查詢低極高flatMap 配合限流4.2 規(guī)則二看內(nèi)部流的類型是 Mono 還是 Flux這里有一個(gè)很多教程沒強(qiáng)調(diào)的細(xì)節(jié)扁平化操作符的映射函數(shù)返回的是 Publisher具體是 Mono 還是 Flux會(huì)影響你對(duì)并發(fā)和順序的預(yù)期。內(nèi)部是 Mono 時(shí)每個(gè)內(nèi)部流只會(huì)發(fā)出一個(gè)元素就結(jié)束flatMap 和 concatMap 的順序差異主要體現(xiàn)在第一個(gè)結(jié)果何時(shí)出現(xiàn)而不會(huì)出現(xiàn)一個(gè)內(nèi)部流的多個(gè)元素被另一個(gè)流打斷的情況。內(nèi)部是 Flux 時(shí)就不一樣了一個(gè)內(nèi)部流可能連續(xù)發(fā)出多個(gè)元素flatMap 會(huì)把不同流的元素交錯(cuò)合并順序問題會(huì)被放大。我之前排查過一個(gè)詭異的現(xiàn)象上游數(shù)據(jù)量很小flatMap 的輸出順序和輸入順序一致同事困惑為什么會(huì)產(chǎn)生亂序 bug。最后發(fā)現(xiàn)真正的差異在內(nèi)部映射函數(shù)某個(gè)分支返回了 Flux另一個(gè)分支返回 Mono不同分支的時(shí)序行為不一致才導(dǎo)致線上偶發(fā)亂序。所以寫映射函數(shù)時(shí)明確返回值是 Mono 還是 Flux對(duì)你的心智負(fù)擔(dān)影響很大。4.3 規(guī)則三用并發(fā)度參數(shù)保護(hù)下游資源不管選擇 flatMap 還是 flatMapSequential我都建議顯式設(shè)置并發(fā)度別依賴 256 這個(gè)默認(rèn)值。256 這個(gè)數(shù)字對(duì)大部分內(nèi)部操作來說都太大了尤其是你的內(nèi)部操作要訪問數(shù)據(jù)庫、調(diào)用第三方HTTP接口時(shí)瞬間 256 路并發(fā)打出去沒有幾個(gè)外部服務(wù)能穩(wěn)穩(wěn)接住。我之前有一個(gè)數(shù)據(jù)同步任務(wù)上游是幾萬條變更記錄每條變更要去查一個(gè)老系統(tǒng)的接口。最開始沒做并發(fā)限制上線第一天就把老系統(tǒng)的連接池打崩了對(duì)方直接拒絕后續(xù)請(qǐng)求。后來改成 flatMap(..., 32)并加了重試和退避整個(gè)任務(wù)從頻繁超時(shí)變成穩(wěn)定跑完。限制并發(fā)的本質(zhì)是控制你對(duì)別人的沖擊程度這和響應(yīng)式本身沒關(guān)系但操作符給了你這個(gè)旋鈕不用就是浪費(fèi)。另外如果并發(fā)度設(shè)得太低比如 4而每個(gè)內(nèi)部流又要做網(wǎng)絡(luò)交互整體吞吐可能還不如串行優(yōu)化得好。并發(fā)度設(shè)置不是一個(gè)固定值要根據(jù)下游服務(wù)的響應(yīng)時(shí)間、連接池大小、單次任務(wù)的體量綜合估算。4.4 flatMapSequential既要并發(fā)又要保序很多人會(huì)在 flatMap 和 concatMap 之間糾結(jié)其實(shí) Reactor 還給了第三個(gè)選項(xiàng)flatMapSequential。它結(jié)合了兩邊的優(yōu)點(diǎn)內(nèi)部流是并發(fā)執(zhí)行的但最終發(fā)射到下游的元素會(huì)按照上游輸入的順序重新排列。用法很簡(jiǎn)單Flux.just(u-1, u-2, u-3) .flatMapSequential(this::fetchDetail)它的內(nèi)部實(shí)現(xiàn)會(huì)為每個(gè)元素的結(jié)果保留一個(gè)槽位哪個(gè)內(nèi)部流先完成先放進(jìn)屬于自己的槽位但發(fā)射動(dòng)作必須按順序來第一個(gè)槽位沒有結(jié)果后面的結(jié)果就算先到也得等待。flatMapSequential 適合報(bào)表導(dǎo)出這類場(chǎng)景既希望大量查詢并發(fā)跑又必須保證輸出順序與輸入順序一致。我自己的經(jīng)驗(yàn)是只要并發(fā)安全和順序要求同時(shí)存在無腦選 flatMapSequential 大概率比 concatMap 合適因?yàn)樗鼪]有完全丟掉并發(fā)收益。但它也有代價(jià)每個(gè)內(nèi)部流的生命周期變長(zhǎng)因?yàn)橄韧瓿傻慕Y(jié)果要緩存在槽位里等前序結(jié)果完成才能發(fā)出去內(nèi)存占用比 flatMap 和 concatMap 高一些。數(shù)據(jù)量極大時(shí)要注意內(nèi)存水位。5. 我踩過的幾個(gè)扁平化相關(guān)的坑5.1 在 flatMap 里用了阻塞調(diào)用背壓和調(diào)度全廢了剛開始寫響應(yīng)式時(shí)最容易犯的錯(cuò)就是在 flatMap 的映射函數(shù)里直接寫阻塞代碼.flatMap(id - { String result jdbcTemplate.queryForObject(...); // 阻塞調(diào)用 return Mono.just(result); })從語法上說代碼能跑但語義上完全跑偏了。flatMap 的映射函數(shù)可能在事件循環(huán)線程上被回調(diào)一個(gè)線程池就那么幾個(gè)線程你在里面 sleep 或做 JDBC 同步查詢等于把整個(gè)并發(fā)度卡死。更麻煩的是背壓機(jī)制也被繞過因?yàn)槟銢]有通過異步信號(hào)告知下游當(dāng)前處理的進(jìn)度。正確做法是把阻塞調(diào)用包裝成異步比如使用專門的阻塞適配器或者提前把數(shù)據(jù)查好再進(jìn)入響應(yīng)式流。如果實(shí)在繞不開至少要換一個(gè)足夠大的彈性線程池避免占住事件循環(huán)線程。5.2 內(nèi)部流為空時(shí)flatMap 會(huì)靜默跳過有一個(gè)很容易忽略的邊界情況映射函數(shù)返回的內(nèi)部流可能一個(gè)元素都不發(fā)直接就結(jié)束了。這時(shí)候 flatMap 的行為是默默跳過這個(gè)元素繼續(xù)處理其他內(nèi)部流。它不會(huì)報(bào)錯(cuò)也不會(huì)中斷。這種設(shè)計(jì)本身沒問題但容易掩蓋邏輯 bug。有一次我做一個(gè)用戶存檔遷移上游是用戶ID內(nèi)部流去查該用戶的歷史訂單新用戶沒訂單內(nèi)部流自然是空的。結(jié)果下游匯總時(shí)發(fā)現(xiàn)用戶數(shù)少了排查半天才發(fā)現(xiàn)是空流被靜默過濾掉了。如果你確實(shí)需要感知某些元素沒有結(jié)果可以在內(nèi)部流里加上默認(rèn)值.flatMap(id - fetchOrders(id) .defaultIfEmpty(new EmptyOrder(id)))這樣每個(gè)用戶至少會(huì)產(chǎn)出一個(gè)元素下游更容易判斷哪些是真正的空結(jié)果。5.3 內(nèi)部流異常會(huì)直接拖垮整個(gè)流程flatMap 的另一個(gè)默認(rèn)行為是內(nèi)部流發(fā)出 error 信號(hào)后外層流會(huì)立刻終止。你在一個(gè)內(nèi)部流里的下游服務(wù)超時(shí)可能導(dǎo)致整批數(shù)據(jù)處理中斷而不是只跳過那一條。這個(gè)行為和 Java 8 Stream 里一個(gè)元素拋異常導(dǎo)致整個(gè)流失敗很像。響應(yīng)式編程本身也是這個(gè)哲學(xué)錯(cuò)誤用信號(hào)表達(dá)而非吞掉。所以如果你的業(yè)務(wù)要求單條失敗不影響整體必須在內(nèi)部流里先做錯(cuò)誤處理例如用 onErrorResume 把異常轉(zhuǎn)成一個(gè)結(jié)果而不是把錯(cuò)誤拋到外層。我一般會(huì)在扁平化操作之前做一個(gè)錯(cuò)誤兜底策略.flatMap(id - fetchDetail(id) .onErrorResume(e - Mono.just(defaultDetail(id))))這樣單個(gè)查詢失敗只會(huì)出來一個(gè)默認(rèn)結(jié)果外層流能繼續(xù)跑完。5.4 concatMap 在上游快、下游慢時(shí)也會(huì)內(nèi)存膨脹concatMap 看起來是串行的好像不會(huì)堆積太多數(shù)據(jù)但其實(shí)它也有預(yù)取隊(duì)列。上游發(fā)射速度遠(yuǎn)大于處理速度時(shí)上游還沒來得及發(fā)完的上萬個(gè)元素會(huì)源源不斷進(jìn)入緩沖區(qū)。等待處理的數(shù)據(jù)量越大內(nèi)存占用就越高。我遇到過一次 OOM場(chǎng)景是從數(shù)據(jù)庫一次性查了五萬條記錄下游對(duì)每一條做一次遠(yuǎn)程調(diào)用用了 concatMap 保序。遠(yuǎn)程調(diào)用平均要幾百毫秒五萬條全部處理完成需要很長(zhǎng)時(shí)間而數(shù)據(jù)庫端的數(shù)據(jù)還在往上游流里灌背壓沒能把內(nèi)存壓住最終堆內(nèi)存被打爆。遇到這種場(chǎng)景我的建議是先把數(shù)據(jù)分批控制上游發(fā)射速度或者在 concatMap 前后搭配 limitRate 進(jìn)行限速。另一個(gè)思路是改用 flatMapSequential 配合合適的并發(fā)度至少內(nèi)部流并發(fā)執(zhí)行能更快消化隊(duì)列降低堆積風(fēng)險(xiǎn)。5.5 調(diào)試順序問題用 log 觀察沒那么玄如果你遇到了疑似亂序的 bug最有效的工具其實(shí)是 log 操作符。它會(huì)在每個(gè)事件發(fā)生時(shí)打印 onNext、onComplete、onSubscribe 等信號(hào)配合元素內(nèi)容你能直觀看到發(fā)射順序和耗時(shí)。Flux.just(u-1, u-2, u-3) .flatMap(this::fetchName) .log(flat-map-demo) .subscribe();跑一次之后把日志時(shí)間線拉出來flatMap 和 concatMap 的差異一目了然。我個(gè)人在寫復(fù)雜響應(yīng)式鏈路時(shí)習(xí)慣先在每個(gè)操作符之間臨時(shí)加 log等流程穩(wěn)定后再刪掉。這個(gè)習(xí)慣幫我定位了很多看起來玄學(xué)的順序怪問題。6. 一句話口訣和兩個(gè)想推薦的練習(xí)如果看完這篇只記住一件事那我希望是這句順序重要用 concatMap性能優(yōu)先用 flatMap既要順序又要并發(fā)用 flatMapSequential。這句話我每次評(píng)審響應(yīng)式代碼時(shí)都會(huì)在腦子里過一遍基本能解決絕大多數(shù)選型問題。最后給你兩個(gè)可以動(dòng)手驗(yàn)證的小練習(xí)練完對(duì)扁平化的理解會(huì)比讀十篇博客都有用。第一個(gè)練習(xí)模擬一個(gè)用戶列表轉(zhuǎn)用戶詳情列表的任務(wù)給不同用戶設(shè)置不同的延遲時(shí)間分別用 flatMap、concatMap、flatMapSequential 跑三遍打印輸出順序。你會(huì)在控制臺(tái)直觀看到三種操作符的時(shí)序差異。第二個(gè)練習(xí)在你自己的項(xiàng)目里找一個(gè)目前用 map 處理集合數(shù)據(jù)的響應(yīng)式接口試著把它改成 flatMap 并加好并發(fā)度參數(shù)觀察下游壓力和耗時(shí)變化。只要不做順序敏感的業(yè)務(wù)大概率能發(fā)現(xiàn)明顯吞吐提升。我在實(shí)際使用中的一個(gè)體會(huì)是flatMap 和 concatMap 之所以容易讓人困惑不是因?yàn)?API 復(fù)雜而是它們迫使你直面順序和并發(fā)不可兼得這個(gè)真實(shí)約束。在熟悉這兩個(gè)極端之后再去理解 flatMapSequential、switchMap 這些高階變體整個(gè)知識(shí)體系就串起來了。也建議你在下次遇到亂序 bug 時(shí)先別急著把 flatMap 換成 concatMap仔細(xì)想想業(yè)務(wù)是否真的完全串行也許 flatMapSequential 這個(gè)中間選項(xiàng)才是更劃算的那一個(gè)。