深潜Kotlin协程（二十一）：Flow 生命周期函数

系列电子书：传送门

Flow 可以想象成一个管道，请求的值在一个方向上流动，而相应产生的值在另一个方向上流动。当 flow 完成或出现异常时，这些信息也会被传递，并关闭途中的中间步骤。因此，当这些值开始流动时，我们可以监听值、异常或其它特征事件（如开始或完成）。为此，我们使用了 onEach、 onStart、onCompletion、onEmpty 和 catch 等方法。下面让我逐一解释这些生命周期方法。

onEach

为了响应每个流动的值，我们使用 onEach 函数。

suspend fun main() {flowOf(1, 2, 3, 4).onEach { print(it) }.collect() // 1234
}

onEach 的 lambda 表达式是挂起的，元素依次按顺序（顺序）被处理。因此，如果我们在 onEach 添加 delay 函数，我们将延迟每个值的流动。

suspend fun main() {flowOf(1, 2).onEach { delay(1000) }.collect { println(it) }
}
// (1 sec)
// 1
// (1 sec)
// 2

onStart

onStart 函数设置一个监听器，一旦 flow 启动，就会回调该监听器。需要注意的是， onStart 并不会等待第一个元素的响应，而是当我们请求第一个元素时，它就会被调用。

suspend fun main() {flowOf(1, 2).onEach { delay(1000) }.onStart { println("Before") }.collect { println(it) }
}
// Before
// (1 sec)
// 1
// (1 sec)
// 2

在 onStart（以及在 onCompletion、onEmpty、catch）中可以发射元素，这些元素将该处往下流动。

suspend fun main() {flowOf(1, 2).onEach { delay(1000) }.onStart { emit(0) }.collect { println(it) }
}
// 0
// (1 sec)
// 1
// (1 sec)
// 2

onCompletion

有几种方法可以完成一个 flow。最常见的是在 flow 构建器完成时（比如发送了最后一个元素），尽管有可能会出现在未捕获异常或者协程取消的情况下。在所有这些情况下，我们都可以使用 onCompletion 方法为 flow 的完成添加一个监听器。

suspend fun main() = coroutineScope {flowOf(1, 2).onEach { delay(1000) }.onCompletion { println("Completed") }.collect { println(it) }
}
// (1 sec)
// 1
// (1 sec)
// 2
// Completedsuspend fun main() = coroutineScope {val job = launch {flowOf(1, 2).onEach { delay(1000) }.onCompletion { println("Completed") }.collect { println(it) }}delay(1100)job.cancel()
}
// (1 sec)
// 1
// (0.1 sec)
// Completed

在 Android 中，我们经常使用 onStart 来展示进度条（等待网络响应的指示器），之后我们使用 onCompletion 来隐藏它。

fun updateNews() {scope.launch {newsFlow().onStart { showProgressBar() }.onCompletion { hideProgressBar() }.collect { view.showNews(it) }}
}

onEmpty

flow 可能在不发射任何值的情况下完成，有可能是出现意外状况。对于这种情况，有一个 onEmpty 函数，它在 flow 完成但没有发出任何元素时会回调。我们可以使用 onEmpty 来发射一些默认值。

suspend fun main() = coroutineScope {flow<List<Int>> { delay(1000) }.onEmpty { emit(emptyList()) }.collect { println(it) }
}
// (1 sec)
// []

catch

在 flow 构建器或处理值的任何时刻，都可能发生异常。这样的异常会向下流动，关闭途中每个处理步骤；然而，它是可以被捕获和管理的。为此，我们可以使用 catch 方法。这个监听器接收异常作为参数，并允许你执行恢复操作。

class MyError : Throwable("My error")val flow = flow {emit(1)emit(2)throw MyError()
}suspend fun main(): Unit {flow.onEach { println("Got $it") }.catch { println("Caught $it") }.collect { println("Collected $it") }
}
// Got 1
// Collected 1
// Got 2
// Collected 2
// Caught MyError: My error

在上面的例子中，onEach 没有对异常做出响应。同样的情况也发生在其他功能上，如 map、 filter 等。只有 onCompletion 的处理才会被调用。

catch 方法通过捕获来阻止异常的传播。虽然前面的步骤已经完成了，但是 catch 仍然可以发射出新的值，并保持 flow 的其余部分处于活跃状态。

val flow = flow {emit("Message1")throw MyError()
}suspend fun main(): Unit {flow.catch { emit("Error") }.collect { println("Collected $it") }
}
// Collected Message1
// Collected Error

catch 只会对上游定义的函数中抛出的异常做出响应（可以想象，当下流还有异常时，仍需要捕获异常）。

在 Android 中，我们经常使用 catch 来展示 flow 中发生的异常：

fun updateNews() {scope.launch {newsFlow().catch { view.handleError(it) }.onStart { showProgressBar() }.onCompletion { hideProgressBar() }.collect { view.showNews(it) }}
}

我们也可以使用 catch 来发出默认数据以显示在屏幕上，比如空列表。

fun updateNews() {scope.launch {newsFlow().catch {view.handleError(it)emit(emptyList())}.onStart { showProgressBar() }.onCompletion { hideProgressBar() }.collect { view.showNews(it) }}
}

未捕获的异常

flow 中出现未捕获的异常则会立即取消该 flow，并且 collect 会重新抛出此异常。这种行为是挂起函数的典型行为， coroutineScope 也有同样的行为，典型的应对方法是使用 try-catch 块在 flow 的外部捕获异常：

val flow = flow {emit("Message1")throw MyError()
}suspend fun main(): Unit {try {flow.collect { println("Collected $it") }} catch (e: MyError) {println("Caught")}
}
// Collected Message1
// Caught

请注意，使用 catch 并不能防止终端操作中出现异常（因为 catch 不能用最后一个操作之后）。因此，如果 collect 中有一个异常，它将不会捕获，而是将抛出一个错误。

val flow = flow {emit("Message1")emit("Message2")
}suspend fun main(): Unit {flow.onStart { println("Before") }.catch { println("Caught $it") }.collect { throw MyError() }
}
// Before
// Exception in thread "..." MyError: My error

因此，通常的做法是将逻辑层从 collect 移动到 onEach，并将其放在 catch 之前。当我们怀疑 collect 的逻辑可能会出现异常时，这样做会特别有用。如果我们将逻辑操作从 collect 中挪走，就可以确定 catch 能捕获所有的异常。

val flow = flow {emit("Message1")emit("Message2")
}
suspend fun main(): Unit {flow.onStart { println("Before") }.onEach { throw MyError() }.catch { println("Caught $it") }.collect()
}
// Before
// Caught MyError: My error

flowOn

传给 flow 操作（如 onEach、onStart、onCompletion 等）的 lambda 表达式及其构建器（如 flow {..} 或 channelFlow{..}）都是挂起的。挂起函数需要有一个上下文，并且应该与它们的父协程相关联（结构化并发）。因此，你可能想知道这些函数的上下文都来自哪里。答案是：从调用的 collect 的上下文中而来。

fun usersFlow(): Flow<String> = flow {repeat(2) {val ctx = currentCoroutineContext()val name = ctx[CoroutineName]?.nameemit("User$it in $name")}
}suspend fun main() {val users = usersFlow()withContext(CoroutineName("Name1")) {users.collect { println(it) }}withContext(CoroutineName("Name2")) {users.collect { println(it) }}
}
// User0 in Name1
// User1 in Name1
// User0 in Name2
// User1 in Name2

这段代码是如何工作的？终端操作会调用来自上游的元素，从而提供协程上下文。然而，它也可以通过 flowOn 函数进行修改（修改协程上下文）：

suspend fun present(place: String, message: String) {val ctx = coroutineContextval name = ctx[CoroutineName]?.nameprintln("[$name] $message on $place")
}fun messagesFlow(): Flow<String> = flow {present("flow builder", "Message")emit("Message")
}suspend fun main() {val users = messagesFlow()withContext(CoroutineName("Name1")) {users.flowOn(CoroutineName("Name3")).onEach { present("onEach", it) }.flowOn(CoroutineName("Name2")).collect { present("collect", it) }}
}
// [Name3] Message on flow builder
// [Name2] Message on onEach
// [Name1] Message on collect

请记住， flowOn 只适用于 flow 中位于上游的函数。

launchIn

collect 是一个挂起函数，它会挂起一个协程直到 flow 完成。我们通常会使用 launch 构建器对其进行包装，以便 flow 处理可以在另一个协程上启动。为了优化这种情况，有一个 launchIn 函数，它会在作为参数传递的作用域上调用 collect。

fun <T> Flow<T>.launchIn(scope: CoroutineScope): Job =scope.launch { collect() }

launchIn 通常用来在一个单独的协程中启动一个 flow。

suspend fun main(): Unit = coroutineScope {flowOf("User1", "User2").onStart { println("Users:") }.onEach { println(it) }.launchIn(this)
}
// Users:
// User1
// User2

总结

在本章中，我们学习了不同的 flow 功能。现在我们知道如何在 flow 开始时、结束时或者在每个元素上执行某些操作；我们还知道如何捕获异常，以及如何在新的协程中启动 flow。这些都是被广泛使用的基本工具，特别是在 Android 开发中。例如，下面是一段 Android 中使用 flow 的代码：

fun updateNews() {newsFlow().onStart { showProgressBar() }.onCompletion { hideProgressBar() }.onEach { view.showNews(it) }.catch { view.handleError(it) }.launchIn(viewModelScope)
}