中間運算子

為什麼需要中間運算子#

Flow 是冷資料流（Cold Stream），在 collect 之前不會執行；而真正讓 Flow 強大的關鍵之一，是我們可以在 collect 之前先用一些函式對資料做轉換。這些函式被稱為 中間運算子（Intermediate Operator）。

中間運算子的概念與 Functional Programming 一致：在輸出之前對資料做一連串的處理，把舊的 Flow 轉成新的 Flow。熟悉 FP 的開發者對這些名稱應該都不會陌生。

延續一個簡單的範例：

fun flow(): Flow<Int> = flow {
    println("Flow started")
    repeat(10) {
        delay(100)
        emit(it)
    }
}

fun main() = runBlocking {
    val flow = flow()
    flow.collect { value -> println(value) }
}

預設執行會印出 0 ~ 9。下面套用幾個常見的中間運算子。

map#

map 把每個元素映射到另一個值（或型別），常用來做資料轉換：

inline fun <T, R> Flow<T>.map(crossinline transform: suspend (T) -> R): Flow<R>

fun main() = runBlocking {
    val flow = flow()
    flow.map { it * it }
        .collect { println(it) }
}

Flow started
0
1
4
9
16
25
36
49
64
81

filter#

filter 顧名思義是篩選資料，傳入一個 lambda 回傳 Boolean，true 才會往下流：

inline fun <T> Flow<T>.filter(crossinline predicate: suspend (T) -> Boolean): Flow<T>

把偶數過濾出來：

fun main() = runBlocking {
    val flow = flow()
    flow.filter { it % 2 == 0 }
        .collect { println(it) }
}

Flow started
0
2
4
6
8

take#

take 只保留前面指定數量的元素。如果指定的數量超過 Flow 的元素數，會以實際數量為準；傳入負值會丟出 java.lang.IllegalArgumentException。

fun <T> Flow<T>.take(count: Int): Flow<T>

取前 3 個：

fun main() = runBlocking {
    val flow = flow()
    flow.take(3)
        .collect { println(it) }
}

Flow started
0
1
2

take 是從 Flow 的最前面開始取。

zip#

zip 把兩個 Flow 一對一組合起來：

fun <T1, T2, R> Flow<T1>.zip(other: Flow<T2>, transform: suspend (T1, T2) -> R): Flow<R>

fun main() = runBlocking {
    val stringFlow = listOf("a", "b", "c", "d").asFlow()
    val intFlow = (1..3).asFlow()
    intFlow.zip(stringFlow) { int, string -> "$int-$string" }
        .collect { println(it) }
}

1-a
2-b
3-c

zip 結合的數量會等於兩條 Flow 中比較短的那條，多出來的元素會被丟掉。

combine#

combine 也是用來組合多條 Flow，但行為與 zip 不同：

fun main() = runBlocking {
    val stringFlow = listOf("a", "b", "c", "d").asFlow()
    val intFlow = (1..3).asFlow()
    intFlow.combine(stringFlow) { int, string -> "$int-$string" }
        .collect { println(it) }
}

1-a
2-b
3-c
3-d

差異：

運算子	以哪條 Flow 為準	缺少對應元素時	可組合條數
`zip`	以較短的那條為準	多餘元素會被捨棄	只能組兩條
`combine`	以較長的那條為準	會用前一個值補上	支援多條

combine 支援多條 Flow，函式多載如下：

fun <T1, T2, R> Flow<T1>.combine(flow: Flow<T2>, transform: suspend (T1, T2) -> R): Flow<R>
fun <T1, T2, R> combine(flow: Flow<T1>, flow2: Flow<T2>, transform: suspend (T1, T2) -> R): Flow<R>
fun <T1, T2, T3, R> combine(flow: Flow<T1>, flow2: Flow<T2>, flow3: Flow<T3>, transform: suspend (T1, T2, T3) -> R): Flow<R>
fun <T1, T2, T3, T4, R> combine(flow: Flow<T1>, flow2: Flow<T2>, flow3: Flow<T3>, flow4: Flow<T4>, transform: suspend (T1, T2, T3, T4) -> R): Flow<R>
fun <T1, T2, T3, T4, T5, R> combine(flow: Flow<T1>, flow2: Flow<T2>, flow3: Flow<T3>, flow4: Flow<T4>, flow5: Flow<T5>, transform: suspend (T1, T2, T3, T4, T5) -> R): Flow<R>
inline fun <T, R> combine(vararg flows: Flow<T>, crossinline transform: suspend (Array<T>) -> R): Flow<R>
inline fun <T, R> combine(flows: Iterable<Flow<T>>, crossinline transform: suspend (Array<T>) -> R): Flow<R>

串接多個中間運算子#

中間運算子可以一個接一個串起來，每經過一層就會產生新的 Flow，直到最後的 collect：

fun main() = runBlocking {
    val flow = flow()
    flow.map { it * 3 }
        .filter { it % 2 == 0 }
        .take(2)
        .collect { println(it) }
}

Flow started
0
6

從輸出只有一個 Flow started 可以看出：原始的 Flow 只會被執行一次，每個中間運算子都是「在這條鏈上多包一層」，最終由 collect 觸發。

其他常見運算子#

除了上面介紹的，還有幾個值得認識的中間運算子（後續章節會視情況提到）：

transform：比 map 更彈性，可以在內部多次 emit，把一個元素拆成多個輸出。
buffer：把 emit 與 collect 解耦，讓上游可以先把資料堆進緩衝區。
debounce：在固定時間內沒有新元素才把最後一個放出去，常用在輸入抖動。
throttle：依時間節流，控制單位時間內的元素數量。
conflate：類似 Conflated Channel，下游來不及收時舊值會被覆蓋。

小結#

中間運算子讓資料在 collect 之前可以被一連串轉換。
它們的命名與 Functional Programming 一致：map、filter、take、zip、combine 等。
中間運算子可以任意組合，但都不會自己觸發資料流動，必須由終端運算子（Terminal Operator）啟動。

原文出處#

原書/iThome：https://ithelp.ithome.com.tw/articles/10270555