19. 處理最新值

當(dāng)發(fā)射器和收集器都很慢的時(shí)候，合并是加快處理速度的一種方式。它通過(guò)刪除發(fā)射值來(lái)實(shí)現(xiàn)材部。 另一種方式是取消緩慢的收集器，并在每次發(fā)射新值的時(shí)候重新啟動(dòng)它唯竹。有一組與 xxx 操作符執(zhí)行相同基本邏輯的 xxxLatest 操作符乐导，但是在新值產(chǎn)生的時(shí)候取消執(zhí)行其塊中的代碼。讓我們?cè)谙惹暗氖纠袊L試更換 [conflate]

val time = measureTimeMillis {
    simple()
        .collectLatest { value -> // 取消并重新發(fā)射最后一個(gè)值
            println("Collecting $value") 
            delay(300) // 假裝我們花費(fèi) 300 毫秒來(lái)處理它
            println("Done $value") 
        } 
}   
println("Collected in $time ms")

Collecting 1
Collecting 2
Collecting 3
Done 3
Collected in 680 ms

20. 組合多個(gè)流

Zip

就像 Kotlin 標(biāo)準(zhǔn)庫(kù)中的 Sequence.zip 擴(kuò)展函數(shù)一樣摩窃， 流擁有一個(gè) zip 操作符用于組合兩個(gè)流中的相關(guān)值：

val nums = (1..3).asFlow() // 數(shù)字 1..3
val strs = flowOf("one", "two", "three") // 字符串
nums.zip(strs) { a, b -> "$a -> $b" } // 組合單個(gè)字符串
    .collect { println(it) } // 收集并打印
1 -> one
2 -> two
3 -> three

Combine

當(dāng)流表示一個(gè)變量或操作的最新值時(shí)兽叮，可能需要執(zhí)行計(jì)算，這依賴于相應(yīng)流的最新值猾愿，并且每當(dāng)上游流產(chǎn)生值的時(shí)候都需要重新計(jì)算鹦聪。這種相應(yīng)的操作符家族稱為 [combine]

例如，先前示例中的數(shù)字如果每 300 毫秒更新一次蒂秘，但字符串每 400 毫秒更新一次泽本， 然后使用 操作符合并它們，但仍會(huì)產(chǎn)生相同的結(jié)果姻僧， 盡管每 400 毫秒打印一次結(jié)果：

val nums = (1..3).asFlow().onEach { delay(300) } // 發(fā)射數(shù)字 1..3规丽，間隔 300 毫秒
val strs = flowOf("one", "two", "three").onEach { delay(400) } // 每 400 毫秒發(fā)射一次字符串
val startTime = System.currentTimeMillis() // 記錄開始的時(shí)間
nums.zip(strs) { a, b -> "$a -> $b" } // 使用“zip”組合單個(gè)字符串
    .collect { value -> // 收集并打印
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    } 

1 -> one at 427 ms from start
2 -> two at 827 ms from start
3 -> three at 1229 ms from start

然而，當(dāng)在這里使用 [combine]操作符來(lái)替換 [zip]

val nums = (1..3).asFlow().onEach { delay(300) } // 發(fā)射數(shù)字 1..3撇贺，間隔 300 毫秒
val strs = flowOf("one", "two", "three").onEach { delay(400) } // 每 400 毫秒發(fā)射一次字符串
val startTime = System.currentTimeMillis() // 記錄開始的時(shí)間
nums.combine(strs) { a, b -> "$a -> $b" } // 使用“combine”組合單個(gè)字符串
    .collect { value -> // 收集并打印
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 

}
1 -> one at 452 ms from start
2 -> one at 651 ms from start
2 -> two at 854 ms from start
3 -> two at 952 ms from start
3 -> three at 1256 ms from start

21. 展平流

流表示異步接收的值序列赌莺，所以很容易遇到這樣的情況： 每個(gè)值都會(huì)觸發(fā)對(duì)另一個(gè)值序列的請(qǐng)求。比如說(shuō)松嘶，我們可以擁有下面這樣一個(gè)返回間隔 500 毫秒的兩個(gè)字符串流的函數(shù)

fun requestFlow(i: Int): Flow<String> = flow {
    emit("$i: First") 
    delay(500) // 等待 500 毫秒
    emit("$i: Second")    
}

現(xiàn)在艘狭，如果我們有一個(gè)包含三個(gè)整數(shù)的流，并為每個(gè)整數(shù)調(diào)用 requestFlow翠订，如下所示：

(1..3).asFlow().map { requestFlow(it) }

然后我們得到了一個(gè)包含流的流（Flow<Flow<String>>）巢音，需要將其進(jìn)行展平為單個(gè)流以進(jìn)行下一步處理。集合與序列都擁有 [flatten]與 [flatMap] 操作符來(lái)做這件事尽超。然而官撼，由于流具有異步的性質(zhì)，因此需要不同的展平模式似谁， 為此傲绣，存在一系列的流展平操作符。
flatMapConcat

連接模式由 [flatMapConcat]與 [flattenConcat]操作符實(shí)現(xiàn)棘脐。它們是相應(yīng)序列操作符最相近的類似物斜筐。它們?cè)诘却齼?nèi)部流完成之前開始收集下一個(gè)值，如下面的示例所示：

al startTime = System.currentTimeMillis() // 記錄開始時(shí)間
(1..3).asFlow().onEach { delay(100) } // 每 100 毫秒發(fā)射一個(gè)數(shù)字 
    .flatMapConcat { requestFlow(it) }                                                                           
    .collect { value -> // 收集并打印
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    } 

1: First at 121 ms from start
1: Second at 622 ms from start
2: First at 727 ms from start
2: Second at 1227 ms from start
3: First at 1328 ms from start
3: Second at 1829 ms from start

在輸出中可以清楚地看到 [flatMapConcat] 的順序性質(zhì)：

flatMapMerge

另一種展平模式是并發(fā)收集所有傳入的流蛀缝，并將它們的值合并到一個(gè)單獨(dú)的流顷链，以便盡快的發(fā)射值。 它由 [flatMapMerge] 與 [flattenMerge]操作符實(shí)現(xiàn)屈梁。他們都接收可選的用于限制并發(fā)收集的流的個(gè)數(shù)的 concurrency 參數(shù)（默認(rèn)情況下嗤练，它等于 [DEFAULT_CONCURRENCY]。

val startTime = System.currentTimeMillis() // 記錄開始時(shí)間
(1..3).asFlow().onEach { delay(100) } // 每 100 毫秒發(fā)射一個(gè)數(shù)字 
    .flatMapMerge { requestFlow(it) }                                                                           
    .collect { value -> // 收集并打印
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    } 

1: First at 150 ms from start
2: First at 246 ms from start
3: First at 347 ms from start
1: Second at 651 ms from start
2: Second at 747 ms from start
3: Second at 849 ms from start

[flatMapMerge]的并發(fā)性質(zhì)很明顯：

flatMapLatest

與 collectLatest 操作符類似（在"處理最新值" 小節(jié)中已經(jīng)討論過(guò)）在讶，也有相對(duì)應(yīng)的“最新”展平模式煞抬，在發(fā)出新流后立即取消先前流的收集。 這由 flatMapLatest 操作符來(lái)實(shí)現(xiàn)构哺。

val startTime = System.currentTimeMillis() // 記錄開始時(shí)間
(1..3).asFlow().onEach { delay(100) } // 每 100 毫秒發(fā)射一個(gè)數(shù)字 
    .flatMapLatest { requestFlow(it) }                                                                           
    .collect { value -> // 收集并打印
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    } 

1: First at 225 ms from start
2: First at 357 ms from start
3: First at 458 ms from start
3: Second at 959 ms from start

該示例的輸出很好的展示了 [flatMapLatest]的工作方式：

在一個(gè)新值到來(lái)時(shí)取消了塊中的所有代碼 (本示例中的 { requestFlow(it) }）革答。 這在該特定示例中不會(huì)有什么區(qū)別战坤，由于調(diào)用 requestFlow 自身的速度是很快的，不會(huì)發(fā)生掛起残拐， 所以不會(huì)被取消途茫。然而，如果我們要在塊中調(diào)用諸如 delay 之類的掛起函數(shù)溪食，這將會(huì)被表現(xiàn)出來(lái)囊卜。

總結(jié)Flow 關(guān)鍵字

• asFlow
• flowOf
• flow{...}
• withTimeoutOrNull
• 操作符，take, map,reduce,filter,transform错沃，toList栅组，collect，toSet枢析，first玉掸，single，reduce醒叁，fold排截，
  flowOn,conflate,collectLatest,Combine
  flatMap,flatMapConcat,flatten,flatMapMerge,flatMapLatest