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

Posted 2022-07-09 RikkaTheWorld

系列电子书：传送门

---

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
// Completed

suspend 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]?.name
        emit("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 = coroutineContext
    val name = ctx[CoroutineName]?.name
    println("[$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)

高性能云服务器 精品线路独享带宽，毫秒延迟，年中盛惠 1 折起