終端運算子的角色#

Flow 經過一連串的中間運算子（Intermediate Operator）處理之後，最後一定要透過 終端運算子（Terminal Operator） 把資料送出去；少了終端運算子，整條 Flow 不會被執行。

除了我們最常用的 collect，Flow 還提供其他幾種終端運算子，本篇逐一介紹。

延續一個簡單的範例：

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

collect#

collect 是最常用的終端運算子：

inline suspend fun <T> Flow<T>.collect(crossinline action: suspend (T) -> Unit)
fun main() = runBlocking {
    val flow = flow()
    flow.collect { value -> println(value) }
}

collect 還有一個無參數版本：

suspend fun Flow<*>.collect()

無參數版本沒有回傳值，要怎麼用？通常會搭配 onStart、onEach、onCompletion、catch 這幾個函式（其實它們屬於中間運算子），把所有處理在 collect 之前定義好。

fun main() = runBlocking {
    val flow = flow()
    flow.onEach { println(it) }
        .onCompletion { println("done") }
        .collect()
}

onStart#

在整條 Flow 開始之前先執行一次。不論放在鏈條的哪個位置，都是最先執行：

public fun <T> Flow<T>.onStart(
    action: suspend FlowCollector<T>.() -> Unit
): Flow<T>
fun main() = runBlocking {
    val flow = flow()
    flow.onStart { println("start") }
        .onEach { println(it) }
        .onCompletion { println("done") }
        .collect()
}

不論 onStart 寫在鏈條的哪個位置（前面、後面），輸出結果都會以 start 開頭、done 結尾，中間是 0 ~ 9。

onCompletion#

與 onStart 相反，onCompletion 在 Flow 結束（正常或異常）之後執行：

public fun <T> Flow<T>.onCompletion(
    action: suspend FlowCollector<T>.(cause: Throwable?) -> Unit
): Flow<T>

action 是 FlowCollector<T> 的擴充函式，所以裡面也可以呼叫 emit：

fun main() = runBlocking {
    val flow = flow()
    flow.onStart { println("start") }
        .onEach { println(it) }
        .onCompletion { emit("done") }
        .collect { println(it) }
}

onEach#

走訪 Flow 中的每個元素。注意：擺放位置會影響它接到的內容，因為前面的中間運算子已經改變了 Flow。

public fun <T> Flow<T>.onEach(action: suspend (T) -> Unit): Flow<T>

範例 1：onEach 在 map 前，看到的還是原本的 0 ~ 9：

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

範例 2：onEach 在 map 後，看到的是已經乘以 3 的結果（0、3、6、9、12…27）：

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

catch#

catch 用來捕捉 Flow 在 emit 過程中拋出的例外：

public fun <T> Flow<T>.catch(action: suspend FlowCollector<T>.(Throwable) -> kotlin.Unit): Flow<T>

下面這段刻意丟出例外：

fun main() = runBlocking {
    val flow = flow()
    flow.map { it * 3 }
        .onEach {
            println(it)
            if (it > 10) throw RuntimeException("large than 10")
        }
        .collect()
}

加上 catch 之後，例外就不會中斷整個程式：

fun main() = runBlocking {
    val flow = flow()
    flow.map { it * 3 }
        .onEach {
            println(it)
            if (it > 10) throw RuntimeException("large than 10")
        }
        .catch { println(it) }
        .collect()
}

single#

single 與 collect 相反，只取出一個值：

suspend fun <T> Flow<T>.single(): T

行為：

  • 如果 Flow 沒有任何元素，會拋出 NoSuchElementException。
  • 如果 Flow 多於 1 個元素，會拋出 IllegalStateException。

通常會搭配 take(1) 確保只剩一個元素：

fun main() = runBlocking {
    val flow = flow
    val value = flow
        .map { it * 3 }
        .take(1)
        .single()
    println(value)
}

另外還有 first 終端運算子，行為類似但更寬鬆：只要 Flow 上有元素就回傳第一個，多餘的部分會被忽略。

reduce#

reduce 會把整條 Flow 的元素「漸減」成一個值。每一步同時拿著前一次的累積值與當前值：

suspend fun <S, T : S> Flow<T>.reduce(operation: suspend (S, T) -> S): S

範例：

fun main() = runBlocking {
    val flow = flow
    val reduce = flow
        .reduce { it, it2 ->
            println("$it + $it2")
            it + it2
        }
    println(reduce)
}
Flow started
0 + 1
1 + 2
3 + 3
6 + 4
10 + 5
15 + 6
21 + 7
28 + 8
36 + 9
45

fold#

fold 與 reduce 類似，但多了一個初始值：

inline suspend fun <T, R> Flow<T>.fold(initial: R, crossinline operation: suspend (R, T) -> R): R
fun main() = runBlocking {
    val fold = (1..4).asFlow()
        .fold(1) { it1, it2 -> it1 * it2 }
    println(fold)
}
24

toList#

toList 把 Flow 轉成 Collection：

suspend fun <T> Flow<T>.toList(destination: MutableList<T> = ArrayList()): List<T>
fun main() = runBlocking {
    val reversed = flow().map { it * 3 }
        .catch { println(it) }
        .onCompletion { println("Done") }
        .toList(ArrayList())
        .reversed()
    println(reversed)
}
Flow started
Done
[27, 24, 21, 18, 15, 12, 9, 6, 3, 0]

launchIn#

launchIn 把 Flow 直接交給一個 CoroutineScope 啟動，常見搭配 onEach 使用，避免在外層手動寫一個空 lambda 給 collect。

flow
    .onEach { println(it) }
    .launchIn(scope)

小結#

  • Flow 必須靠終端運算子（Terminal Operator）才會真正執行。
  • 中間運算子可以串很多個，但終端運算子在一條 Flow 上只能呼叫一次。
  • 常見終端運算子包含 collect、single、first、reduce、fold、toList、launchIn 等，依需求選用。

原文出處#

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