Kotlin 协程基础知识总结四 —— Flow

异步流 Flow 主要内容：

1. 认识：特性、构建器与上下文、启动、取消与取消检测特性、缓冲
2. 操作符：过渡操作符、末端操作符、组合、扁平
3. 异常：异常处理、完成

1、认识

1.1 如何异步返回多个值

挂起函数可以异步返回单个值，那如何异步返回多个计算好的值呢？

同步返回多个值

（P56）通过集合、序列、挂起函数返回多个值

先看集合与序列，二者本质上都是同步返回多个值：

	private fun simpleList(): List<Int> = listOf(1, 2, 3)private fun simpleSequence(): Sequence<Int> = sequence {for (i in 1..3) {// sleep() 会阻塞线程（假装在计算），也就是说还是同步的Thread.sleep(1000)yield(i)}}/*** 集合、序列、挂起函数返回多个值*/@Testfun test01() {// 1.集合是同步返回多个值simpleList().forEach { value -> println(value) }// 2.序列确实可以模拟出一段时间返回一个值的情形，但是时间间隔是通过// Thread.sleep() 实现的，本质还是同步的simpleSequence().forEach { value -> println(value) }}

再看挂起函数，配合集合：

	private suspend fun simpleList2(): List<Int> {delay(1000)return listOf(1, 2, 3)}fun test02() = runBlocking<Unit> {// 3.挂起函数加集合，虽然是异步了，但是值是一次性返回的，而不是计算好一个值就立即返回一个值launch {simpleList2().forEach { value -> println(value) }}}

虽然在协程环境中，挂起函数异步返回了多个值，但是这多个值是一起返回的，而不是计算好一个值就立即返回一个值，分多次返回。

是否有在 simpleSequence 的 for 循环中通过执行一个挂起函数，比如 delay 模拟协程计算值的过程，从而实现异步多次返回多个值的想法呢？

想法很好，但并不能实现。因为 sequence 函数的参数 block 是 SequenceScope 的扩展函数：

@SinceKotlin("1.3")
@Suppress("DEPRECATION")
public fun <T> sequence(@BuilderInference block: suspend SequenceScope<T>.() -> Unit): Sequence<T> = Sequence { iterator(block) }

而 SequenceScope 类上被加了 @RestrictsSuspension 注解：

@RestrictsSuspension
@SinceKotlin("1.3")
public abstract class SequenceScope<in T> internal constructor()

这个注解的字面意思就是限制挂起，被注解的类或接口在作为扩展挂起函数的接收者时受到限制。这些挂起扩展只能调用此特定接收者上的其他成员或扩展挂起函数，并受限制不得调用任意的挂起函数：

/*** Classes and interfaces marked with this annotation are restricted when used as receivers for extension* `suspend` functions. These `suspend` extensions can only invoke other member or extension `suspend` functions on this particular* receiver and are restricted from calling arbitrary suspension functions.*/
@SinceKotlin("1.3")
@Target(AnnotationTarget.CLASS)
@Retention(AnnotationRetention.BINARY)
public annotation class RestrictsSuspension

因此，作为被该注解标记的 SequenceScope 的扩展函数 sequence 不能调用 delay 之类的挂起函数，其所能调用的挂起函数有严格的范围限制。

异步返回多个值

（P57）通过 Flow 异步返回多个值

Flow 的使用模式有点像 RxJava：

	private fun simpleFlow() = flow {for (i in 1..3) {// 模拟生产数据的耗时delay(1000)// 发射产生的数据到接收端emit(i)}}@Testfun test03() = runBlocking<Unit> {// 开一个协程用以验证 Flow 是异步的launch {for (i in 1..3) {println("I'm not blocked $i")delay(1500)}}// 4.Flow 是异步，多次返回多个值launch {simpleFlow().collect { value -> println(value) }}}

输出结果：

I'm not blocked 1
1
I'm not blocked 2
2
I'm not blocked 3
3

（P59）Flow 与其他方式的区别，实际上说的是 Flow 的特点：

• 名为 flow 的 Flow 类型构造器函数
• flow{…} 构建块中的代码可以挂起
• simpleFlow() 可以不标 suspend 修饰符
• 流使用 emit 函数发射，collect 函数接收

（P60）流的典型应用，在 Android 中是下载文件：

1.2 流

冷流

（P60）什么是冷流

Flow 是一种类似序列的冷流，flow 构建器中的代码直到流被收集的时候才运行：

	private fun simpleFlow2() = flow {println("Flow started")for (i in 1..3) {// 模拟生产数据的耗时delay(1000)// 发射产生的数据到接收端emit(i)}}@Testfun test04() = runBlocking<Unit> {val flow = simpleFlow2()println("Calling collect...")flow.collect { value -> println(value) }println("Calling collect again...")flow.collect { value -> println(value) }}

输出结果：

Calling collect...
Flow started
1
2
3
Calling collect again...
Flow started
1
2
3

可以看到，是先调用了 collect 之后才运行 flow 的。

在 Kotlin 中，流（Flow）是一种异步数据流的概念，它提供了一种基于协程的回压机制，用于处理异步数据序列。在流的概念中，可以将流分为冷流（Cold Flow）和热流（Hot Flow）两种类型。

1. 冷流（Cold Flow）：
   • 冷流是指每当收集者开始收集数据时，生产者才开始产生数据。换句话说，冷流是惰性的，只有在有收集者订阅时才会启动数据生产。
   • 每个订阅者都会收到自己的数据流，并且订阅者之间不会共享数据。
   • 冷流适用于一对一的数据传输，每个订阅者独立处理数据。
2. 热流（Hot Flow）：
   • 热流是指在数据产生之后，无论是否有收集者订阅，数据都会持续产生。订阅者加入后只能获取到数据流的当前状态，无法获取之前的数据。
   • 热流通常用于广播数据给多个订阅者，或者用于实时事件处理等场景。

在 Kotlin 的流中，使用 flow 构建冷流，而热流通常需要额外的处理，比如使用 SharedFlow、StateFlow 或者其他类似的机制来实现。冷流适用于按需获取数据，而热流适用于实时数据传输和多个订阅者之间共享数据的场景。

冷流和热流的选择取决于具体的使用场景，需要根据数据传输方式、订阅者之间的关系以及数据的实时性要求来选择适合的类型。

流是连续的

（P61）流的连续性

流中的数据是有顺序的，收集时也会按照该数据收集：

	fun test05() = runBlocking<Unit> {(1..5).asFlow().filter { it % 2 == 0 }.map { "string $it" }.collect { println("Collect $it") }}

输出结果：

Collect string 2
Collect string 4

冷流构建器

（P62）流的构建器，常用两种：

1. flowOf 构建器定义一个发射固定值集的流
2. asFlow 扩展函数可以将各种集合与序列转换为流

上述两种构建器都是针对静态数据集使用的，代码示例：

	fun test06() = runBlocking<Unit> {flowOf("one", "two", "three").onEach { delay(1000) }.collect { value -> println(value) }(1..3).asFlow().collect { value -> println(value) }}

输出结果：

one
two
three
1
2
3

流的上下文

（P63）流上下文：

• 流的收集（collect）总是在调用协程的上下文中发生
• 流的操作会继承之前操作中设置的上下文，这种行为被称为上下文保存。这种机制使得在流的不同部分中可以使用不同的上下文，而不必在每个操作符中显式地重新指定上下文
• flow{…} 构建器中的代码必须遵循上下文保存属性，并且不允许从其他上下文中发射（emit）
• flowOn 操作符用于更改流发射（即上流）的上下文
• 在哪里调用 collect，下流的上下文就是哪里，下流上下文决定上流上下文（上流与下流保持一致），如想改变上流的上下文，需使用 flowOn

对流的上下文以及上下文保存的更详尽解释。

**流的上下文：**在 Kotlin 中，流（Flow）可以与协程的上下文（Coroutine Context）一起使用，以控制流的执行环境和调度方式。流的上下文可以影响流的执行方式、线程调度、错误处理等方面。

流的上下文可以通过 flowOn 操作符来指定，用于更改流的执行上下文。例如，可以使用 flowOn(Dispatchers.IO) 来将流的执行调度到 IO 线程池中。这样可以确保流的操作在 IO 线程中执行，避免阻塞主线程。

另外，流的上下文还可以通过 flow { ... }.flowOn(context) 的方式来指定，其中 context 是一个 CoroutineContext 对象。这允许你在流的不同部分中使用不同的上下文，灵活地控制流的执行环境。

流的上下文还与协程作用域相关联。例如，在使用 coroutineScope 或 viewModelScope 等作用域创建流时，流将继承这些作用域的上下文，从而与父协程共享相同的上下文。

**流上下文与协程上下文是否相同：**在 Kotlin 中，流（Flow）的上下文通常是与协程的上下文相关联的，但它们并不完全相同。流的上下文可以控制流的执行环境和调度方式，而协程的上下文则控制整个协程的执行环境。

当创建流时，流的执行环境会受到流上下文的影响。这意味着在流中的每个操作符或者运算符中，都会继承上一个操作符中的上下文，这与协程中的上下文传递方式类似。

流的上下文是通过 flowOn 操作符或者在流构建器中传递的 CoroutineContext 对象来指定的。这些上下文可以用来指定流操作在哪个线程或者调度器中执行。

虽然流的上下文通常与协程的上下文相关联，但是在某些情况下它们可以是不同的。例如，在一个流中，可以通过 flowOn 操作符将流操作切换到不同的线程，而协程的上下文可能并没有被改变。

总的来说，流的上下文是与协程的上下文相关联的，但二者并不完全相同。流的上下文用于控制流的执行环境，而协程的上下文用于控制整个协程的执行环境。在实际使用中，需要根据具体情况来选择合适的上下文来管理流和协程的执行。

**如何理解流的上下文保存：**流的上下文保存指的是在 Kotlin 流（Flow）中，流的操作会继承之前操作中设置的上下文，这种行为被称为上下文保存。这种机制使得在流的不同部分中可以使用不同的上下文，而不必在每个操作符中显式地重新指定上下文。

当在流的操作链中使用 flowOn 操作符或者在流构建器中传递 CoroutineContext 对象时，这些上下文会在流的执行过程中被保存和传递。

理解流的上下文保存的关键点：

1. 上下文传递：

   • 在流的操作中，上一个操作符设置的上下文会被下一个操作符继承。这意味着不需要在每个操作符中重复指定上下文，可以在流的初始部分设置一次即可。

2. 线程切换：

   • 通过在流操作中设置不同的上下文，可以实现在不同线程或者调度器中执行不同的流操作。这种灵活性使得可以根据需要在流的不同部分中进行线程切换。

3. 影响范围：

   • 上下文保存的影响范围通常是在同一个流中。即在同一个流操作链中，上下文会被保存并传递，但是在不同的流实例中上下文是独立的。

4. 协程特性：

   • 流的上下文保存是基于 Kotlin 协程的特性实现的，利用协程的协作和上下文传递机制来确保在流操作中上下文的正确传递和保存。

通过理解流的上下文保存，可以更好地利用流的特性和协程机制，实现灵活的异步流处理和线程调度控制。这种机制使得在复杂的流操作链中管理上下文变得更加便捷和高效。

虽然表面上看起来，流的收集是在协程中，而流的发射并没有在协程中。但实际上，二者是在同一个协程中：

	private fun simpleFlow3() = flow {println("Flow started on ${Thread.currentThread().name}")for (i in 1..3) {// 模拟生产数据的耗时delay(1000)// 发射产生的数据到接收端emit(i)}}@Testfun test07() = runBlocking<Unit> {simpleFlow3().collect { value -> println("Collected $value on ${Thread.currentThread().name}") }}

输出结果：

Flow started on Test worker @coroutine#1
Collected 1 on Test worker @coroutine#1
Collected 2 on Test worker @coroutine#1
Collected 3 on Test worker @coroutine#1

可以看到流的发射与收集是在同一个协程中的。

默认情况下，流的上下文会进行保存，这使得流的发射与收集在同一个线程中，这不太符合我们异步处理问题的思路。比如下载文件应该是在子线程中进行，而下载好的数据要切换中子线程中供 UI 更新所用。可以使用 withContext 切换到 IO 线程吗？答案是不可以：

	private fun simpleFlow4() = flow {withContext(Dispatchers.IO) {println("Flow started on ${Thread.currentThread().name}")for (i in 1..3) {// 模拟生产数据的耗时delay(1000)// 发射产生的数据到接收端emit(i)}}}@Testfun test08() = runBlocking<Unit> {simpleFlow4().collect { value -> println("Collected $value on ${Thread.currentThread().name}") }}

会报异常，说 collect 与 emit 不在同一个协程中：

Flow invariant is violated:Flow was collected in [CoroutineId(1), "coroutine#1":BlockingCoroutine{Active}@16d69f49, BlockingEventLoop@31ec8c42],but emission happened in [CoroutineId(1), "coroutine#1":DispatchedCoroutine{Active}@3fac37fd, Dispatchers.IO].Please refer to 'flow' documentation or use 'flowOn' instead

withContext 不行，但是前面提到过的 flowOn 可以：

	private fun simpleFlow5() = flow {println("Flow started on ${Thread.currentThread().name}")for (i in 1..3) {// 模拟生产数据的耗时delay(1000)// 发射产生的数据到接收端emit(i)}}.flowOn(Dispatchers.IO)@Testfun test09() = runBlocking<Unit> {simpleFlow5().collect { value -> println("Collected $value on ${Thread.currentThread().name}") }}

输出结果：

Flow started on DefaultDispatcher-worker-1 @coroutine#2
Collected 1 on Test worker @coroutine#1
Collected 2 on Test worker @coroutine#1
Collected 3 on Test worker @coroutine#1

Flow 发射是在子线程，收集是在测试的主线程中。

在指定协程中收集流

（P64）在指定的协程中收集流

使用 launchIn 替换 collect 在指定的协程中启动流的收集：

	private fun events() = (1..3).asFlow().onEach { delay(100) }.flowOn(Dispatchers.Default)@Testfun test10() = runBlocking<Unit> {events().onEach { event -> println("Event $event received on ${Thread.currentThread().name}") }.launchIn(CoroutineScope(Dispatchers.IO)).join()}

输出结果：

Event 1 received on DefaultDispatcher-worker-3 @coroutine#2
Event 2 received on DefaultDispatcher-worker-1 @coroutine#2
Event 3 received on DefaultDispatcher-worker-3 @coroutine#2

launchIn 返回的是一个 Job，如果后续想取消还可以调用 cancelAndJoin。

流的取消

（P65）流的取消

流采用与协程同样的协作取消。像往常一样，当流在一个可取消的挂起函数（例如 delay）中被挂起时，流的收集可以被取消。下例的 withTimeoutOrNull() 展示了流是如何在超时的情况下取消并停止执行其代码的：

fun simple(): Flow<Int> = flow { for (i in 1..3) {delay(100)          println("Emitting $i")emit(i)}
}fun main() = runBlocking<Unit> {withTimeoutOrNull(250) { // Timeout after 250ms simple().collect { value -> println(value) } }println("Done")
}

simple() 的流中仅发射两个数字：

Emitting 1
1
Emitting 2
2
Done

（P66）流的取消检测：

• 为方便起见，流构建器对每个发射值执行附加的 ensureActive() 检测以进行取消，这意味着从 flow {…} 发出的繁忙循环是可以取消的
• 处于性能原因，大多数其他流操作不会自行执行其他取消检测。协程繁忙时，必须明确检测是否取消
• 通过 cancellable 操作符执行此操作

先看第一点的验证，使用 flow 构建一个流，emit() 在内部发射数据之前会通过 ensureActive() 检测 Job 是否处于活动状态，如不是则不会发射数据：

	private fun simpleFlow6() = flow {for (i in 1..3) {emit(i)println("Emitting $i")}}@Testfun test11() = runBlocking<Unit> {simpleFlow6().collect { value ->println("$value")if (value == 3) {cancel()}}}

运行结果：

1
Emitting 1
2
Emitting 2
3
Emitting 3BlockingCoroutine was cancelled
kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job="coroutine#1":BlockingCoroutine{Cancelled}@2e8c1c9bat app//kotlinx.coroutines.JobSupport.cancel(JobSupport.kt:1579)at app//kotlinx.coroutines.CoroutineScopeKt.cancel(CoroutineScope.kt:287)at app//kotlinx.coroutines.CoroutineScopeKt.cancel$default(CoroutineScope.kt:285)at app//com.coroutine.basic.CoroutineTest03$test11$1$1.emit(CoroutineTest03.kt:193)

可以看到当下流接收到 3 这个数据调用 cancel 取消流是成功了的。

再看后两点，如果不是 flow {…} 构建的、通过 emit() 发射数据的流，cancel() 是无法取消的：

	fun test12() = runBlocking<Unit> {(1..5).asFlow().collect { value ->println("$value")if (value == 3) {cancel()}}}

运行结果：

1
2
3
4
5BlockingCoroutine was cancelled
kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job="coroutine#1":BlockingCoroutine{Cancelled}@1e6a3214at app//kotlinx.coroutines.JobSupport.cancel(JobSupport.kt:1579)at app//kotlinx.coroutines.CoroutineScopeKt.cancel(CoroutineScope.kt:287)at app//kotlinx.coroutines.CoroutineScopeKt.cancel$default(CoroutineScope.kt:285)at app//com.coroutine.basic.CoroutineTest03$test12$1$1.emit(CoroutineTest03.kt:203)

虽然最后也抛出了取消的异常，但是很明显，取消并不成功，并没有在收到 3 之后就将流停止。

需要在流之后接上 cancellable 操作符才行：

	fun test12() = runBlocking<Unit> {(1..5).asFlow().cancellable().collect { value ->println("$value")if (value == 3) {cancel()}}}

运行结果：

1
2
3BlockingCoroutine was cancelled
kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job="coroutine#1":BlockingCoroutine{Cancelled}@740cae06at app//kotlinx.coroutines.JobSupport.cancel(JobSupport.kt:1579)at app//kotlinx.coroutines.CoroutineScopeKt.cancel(CoroutineScope.kt:287)at app//kotlinx.coroutines.CoroutineScopeKt.cancel$default(CoroutineScope.kt:285)at app//com.coroutine.basic.CoroutineTest03$test12$1$1.emit(CoroutineTest03.kt:204)

1.3 背压与合并

背压：

• buffer()，并发运行流中发射数据的代码
• conflate()，合并发射项，不对每个值进行处理
• collectLatest()，取消并重新发射最后一个值
• 当必须更改 CoroutineDispatcher 时，flowOn 操作符使用了相同的缓冲机制，但是我们在这里显式地请求缓冲而不改变执行上下文

缓冲

（P67）使用缓冲与 flowOn 处理背压

从物理的角度来讲，是指水流受到与流动方向一致的压力称为背压。从程序的角度讲，是指生产者的生产效率大于消费者的消费效率。

现在模拟一个生产者效率大于消费者效率的代码：

	private fun simpleFlow7() = flow {for (i in 1..3) {delay(100)emit(i)println("Emitting $i ${Thread.currentThread().name}")}}@Testfun test13() = runBlocking<Unit> {val time = measureTimeMillis {simpleFlow7().collect { value ->delay(300)println("$value ${Thread.currentThread().name}")}}println("Collected in $time ms")}

假设生产者发射数据需要 100ms，而消费者收集数据需要 300ms，那么发送 3 条数据需要 1200ms 左右：

1 Test worker @coroutine#1
Emitting 1 Test worker @coroutine#1
2 Test worker @coroutine#1
Emitting 2 Test worker @coroutine#1
3 Test worker @coroutine#1
Emitting 3 Test worker @coroutine#1
Collected in 1260 ms

也就是说，当前发射和收集数据是在同一协程中顺序运行的，所以才需要 3 × (100 + 300) = 1200ms 左右的时间。通过使用 buffer 操作符，可以让发射和收集并发执行以解决时间：

	fun test14() = runBlocking<Unit> {// 发射与收集并发执行val time = measureTimeMillis {simpleFlow7().buffer().collect { value ->delay(300)println("$value ${Thread.currentThread().name}")}}println("Collected in $time ms")}

查看运行结果发现发射与收集分别在同一线程的不同协程中执行，时间也有缩短：

Emitting 1 Test worker @coroutine#2
Emitting 2 Test worker @coroutine#2
Emitting 3 Test worker @coroutine#2
1 Test worker @coroutine#1
2 Test worker @coroutine#1
3 Test worker @coroutine#1
Collected in 1097 ms

时间上大概是发射时 delay 的 100ms 加上收集 3 个数据所需的 900ms。

实际上，使用 flowOn 操作符也能达到类似的效果：

	fun test15() = runBlocking<Unit> {val time = measureTimeMillis {simpleFlow7().flowOn(Dispatchers.Default).collect { value ->delay(300)println("$value ${Thread.currentThread().name}")}}println("Collected in $time ms")}

运行结果：

Emitting 1 DefaultDispatcher-worker-1 @coroutine#2
Emitting 2 DefaultDispatcher-worker-1 @coroutine#2
Emitting 3 DefaultDispatcher-worker-1 @coroutine#2
1 Test worker @coroutine#1
2 Test worker @coroutine#1
3 Test worker @coroutine#1
Collected in 1094 ms

需要注意的是，flowOn 操作符实际上使用了相同的缓冲机制，假如要求不改变执行的上下文，那么就只能使用 buffer。

合并与处理最新值

（P68）合并与处理最新值，用到 conflate 和 collectLatest 操作符。

开发过程中有时不需要处理每个值，而只需处理最新的值。此时可使用 conflate 操作符在收集处理缓慢时跳过中间值：

	fun test16() = runBlocking<Unit> {val time = measureTimeMillis {simpleFlow7().conflate().collect { value ->delay(300)println("$value ${Thread.currentThread().name}")}}println("Collected in $time ms")}

运行结果，当第一个数字仍在处理时，第二个和第三个数字已经被发射出来了，因此第二个数字被跳过了，只有最新的数字（第三个数字）被传递给了收集器：

Emitting 1 Test worker @coroutine#2
Emitting 2 Test worker @coroutine#2
Emitting 3 Test worker @coroutine#2
1 Test worker @coroutine#1
3 Test worker @coroutine#1
Collected in 782 ms

处理最新的值使用 collectLatest 操作符：

	fun test17() = runBlocking<Unit> {val time = measureTimeMillis {simpleFlow7().collectLatest { value ->delay(300)println("$value ${Thread.currentThread().name}")}}println("Collected in $time ms")}

运行结果：

Emitting 1 Test worker @coroutine#2
Emitting 2 Test worker @coroutine#2
Emitting 3 Test worker @coroutine#2
3 Test worker @coroutine#5
Collected in 776 ms

2、操作符

函数式编程风格会有很多操作符：

• 可以使用操作符转换流，就像使用集合与序列一样
• 过渡操作符应用于上游流，并返回下游流
• 操作符是冷操作符，就像流一样，这些操作符本身不是挂起函数
• 它们的运行速度很快，返回新的转换流的定义

2.1 转换操作符

（P69）使用 map 操作符：

	private suspend fun performRequest(request: Int): String {delay(1000)return "response $request"}@Testfun test01() = runBlocking<Unit> {(1..3).asFlow().map { value -> performRequest(value) }.collect { value -> println(value) }}

运行结果：

response 1
response 2
response 3

map 内只能进行一次转换，而 transform 内可以进行多次：

    fun test02() = runBlocking<Unit> {(1..3).asFlow().transform { value ->emit("Making request $value")emit(performRequest(value))}.collect { value -> println(value) }}

transform 内可以进行多次转换，通过 emit 发射多个转换后的数据：

Making request 1
response 1
Making request 2
response 2
Making request 3
response 3

2.2 限长操作符

（P70）take 操作符在触及相应限制时会通过抛出异常取消流的执行：

	private fun numbers(): Flow<Int> = flow {try {emit(1)emit(2)println("This line will not execute")emit(3)} finally {println("Finally in numbers")}}fun test03() = runBlocking<Unit> {numbers().take(2).collect { value -> println(value) }}

运行结果：

1
2
Finally in numbers

结果表明，numbers() 中对 flow 函数体的执行在发射出第二个数字后停止。

转换操作符与限长操作符都属于中间流操作符（Intermediate flow operators）。

2.3 末端操作符

（P71）末端流（终止流）操作符是在流上用于启动流收集的挂起函数。collect 是最基础的终止流操作符，其他还有：

• 转化为各种集合，如 toList 与 toSet
• 获取第一个值的操作符 first 与确保发射单个值的操作符 single
• 使用 reduce 与 fold 将流压缩为单个值

以 reduce 为例：

	fun test04() = runBlocking<Unit> {val sum = (1..5).asFlow().map { value -> value * value }.reduce { a, b -> a + b } // 将两个数压缩为一个数，这里采用加法// 输出 1 到 5 的平方和 55println(sum)}

2.4 组合操作符

（P72）组合操作符，zip 和 combine。

就像 Kotlin 标准库中的扩展函数 Sequence.zip 一样，流拥有一个 zip 操作符，用于组合两个流的对应值：

	fun test05() = runBlocking<Unit> {val nums = (1..3).asFlow()val strs = flowOf("one", "two", "three")nums.zip(strs) { a, b -> "$a -> $b" }.collect { println(it) }}

输出：

1 -> one
2 -> two
3 -> three

combine 他没讲……这里参考官方文档的内容引入 combine。假如为 nums 和 strs 两个流的每一个元素发射前加一段延时：

	fun test06() = runBlocking<Unit> {val nums = (1..3).asFlow().onEach { delay(300) }val strs = flowOf("one", "two", "three").onEach { delay(400) }val startTime = System.currentTimeMillis()nums.zip(strs) { a, b -> "$a -> $b" }.collect { value ->println("$value at ${System.currentTimeMillis() - startTime} ms from start")}}

计算结果是一样的，消耗的时间大概是 3 个 400ms：

1 -> one at 438 ms from start
2 -> two at 838 ms from start
3 -> three at 1247 ms from start 

假如被合并的流有新的值发出时，combine 会立即触发组合操作，生成新的值：

	fun test06() = runBlocking<Unit> {val nums = (1..3).asFlow().onEach { delay(300) }val strs = flowOf("one", "two", "three").onEach { delay(400) }val startTime = System.currentTimeMillis()// 如不需要实时响应流中新发射的值就用 zip，否则用 combinenums.combine(strs) { a, b -> "$a -> $b" }.collect { value ->println("$value at ${System.currentTimeMillis() - startTime} ms from start")}}

运行结果：

1 -> one at 467 ms from start
2 -> one at 666 ms from start
2 -> two at 882 ms from start
3 -> two at 975 ms from start
3 -> three at 1286 ms from start

可以看到，两个流中每当有新的数据被发射，combine 就会进行一次组合计算：

• 第一次输出 1 -> one 是在 strs 流延迟 400ms 后计算得到的
• 第二次输出 2 -> one 是 nums 发射第二个数据后，大概是 2 × 300 = 600ms 左右得到的
• 第三次输出 2 -> two 是 strs 发射第二个数据后，大概 2 × 400 = 800ms 左右得到的
• 后续以此类推……

2.5 展平操作符

（P73）流表示异步接收的值序列，因此很容易陷入每个值触发另一个值序列请求的情况。集合和序列具有 flatten 和 flatMap 操作符来进行展平。然而，由于流具有异步的性质，因此需要不同的展平模式，为此，存在一系列的展平流操作符：

• flatMapConcat：连接模式
• flatMapMerge：合并模式
• flatMapLatest：最新展平模式

展平操作符的应用场景就是将嵌套流 Flow<Flow<Data>> 展平为非嵌套的流 Flow<Data>。那何时会产生嵌套流呢？就是前面说的，有请求需要一个流的序列触发另一个值的请求。比如现在有一个流：

	private fun requestFlow(i: Int): Flow<String> = flow {emit("$i: First")delay(500)emit("$i: Second")}

然后另外有一个包含三个整数的流，通过 map 操作符提供 requestFlow 的参数：

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

这样就会得到嵌套流 Flow<Flow<String>> 。下面就看通过三个操作符能达到什么样的效果。

首先是 flatMapConcat：

	fun test07() = runBlocking<Unit> {val startTime = System.currentTimeMillis()(1..3).asFlow().onEach { delay(100) }.flatMapConcat { requestFlow(it) }.collect { value ->println("$value at ${System.currentTimeMillis() - startTime} ms from start")}}

运行结果：

1: First at 134 ms from start
1: Second at 650 ms from start
2: First at 758 ms from start
2: Second at 1269 ms from start
3: First at 1376 ms from start
3: Second at 1891 ms from start

先取出第一个流的第一个元素参与第二个流的运算，然后取出第一个流的第二个元素……内部实际上是先对第一个流做 map 变换然后再做 flattenConcat 操作：

// Merge.kt
@FlowPreview
public fun <T, R> Flow<T>.flatMapConcat(transform: suspend (value: T) -> Flow<R>): Flow<R> =map(transform).flattenConcat()@FlowPreview
public fun <T> Flow<Flow<T>>.flattenConcat(): Flow<T> = flow {collect { value -> emitAll(value) }
}

emitAll() 就是调用流的 collect：

// Collect.kt
public suspend fun <T> FlowCollector<T>.emitAll(flow: Flow<T>) {ensureActive()// Flow 是接口，这个 collect 要看具体实现类了flow.collect(this)
}

再看 flatMapMerge：

	fun test08() = runBlocking<Unit> {val startTime = System.currentTimeMillis()(1..3).asFlow().onEach { delay(100) }.flatMapMerge { requestFlow(it) }.collect { value ->println("$value at ${System.currentTimeMillis() - startTime} ms from start")}}

运行结果：

1: First at 180 ms from start
2: First at 278 ms from start
3: First at 387 ms from start
1: Second at 694 ms from start
2: Second at 787 ms from start
3: Second at 896 ms from start

该结果与 flatMapConcat 相比，顺序不同，时间也节省了一半还要多。这是因为 flatMapMerge 是同时收集所有传入的流，并将它们的值合并到单个流中，以便尽快发射。

需要注意 flatMapMerge 按顺序调用代码块（在此示例中为 { requestFlow(it) }），但并发地收集结果流。这相当于先顺序执行 map { requestFlow(it) } 操作，然后对结果调用 flattenMerge 操作符，实际代码也是这样实现的：

// Merge.kt
@FlowPreview
public fun <T, R> Flow<T>.flatMapMerge(// 用于限制同时收集的并发流的数量，默认值为 16concurrency: Int = DEFAULT_CONCURRENCY,transform: suspend (value: T) -> Flow<R>
): Flow<R> =map(transform).flattenMerge(concurrency)

最后看 flatMapLatest：

	fun test09() = runBlocking<Unit> {val startTime = System.currentTimeMillis()(1..3).asFlow().onEach { delay(100) }.flatMapLatest { requestFlow(it) }.collect { value ->println("$value at ${System.currentTimeMillis() - startTime} ms from start")}}

运行结果：

1: First at 166 ms from start
2: First at 314 ms from start
3: First at 419 ms from start
3: Second at 927 ms from start

结果表明 flatMapLatest 在新流发射时会取消之前收集的流：

• 数字流发射 1 到字符流，requestFlow 发射 “1: First” 之后挂起 500ms，在挂起期间，数字流会发射数字 2，因此 requestFlow 本应在挂起后发射的 “1: Second” 被取消
• 数字流发射 2 在 requestFlow 中转换为 “2: First” 被发射，但是由于 500ms 的挂起，“2: Second” 也会被取消
• 数字流发射 3 在 requestFlow 中转换为 “3: First” 被发射，由于后续数字流不再发射数据了，因此挂起 500ms 后会发射 “3: Second”，不会因为有新的流数据到来而被取消

注意，取消是取消整个代码块：

需要注意的是，当 flatMapLatest 接收到新的值时，它会取消代码块中的所有代码（在此示例中为 { requestFlow(it) }），因此如果在 requestFlow 中使用了挂起函数（如 delay），它们可能会被取消。在这个特定的示例中，requestFlow 调用本身是快速的、非挂起的，并且不能被取消，因此这并不会产生影响。但是，如果我们在 requestFlow 中使用了挂起函数，那么输出中的差异将是可见的。

3、异常处理与完成

3.1 流的异常处理

（P74）流的异常处理：可以通过 try-catch 块或 catch 函数捕获异常。

由于流有发射端和收集端两端，两端都有可能发生异常，因此要在两端考虑异常捕获。首先是收集端：

private fun simple() = flow {for (i in 1..3) {println("Emitting $i")emit(i)}}@Testfun test10() = runBlocking<Unit> {try {simple().collect { value ->println(value)check(value <= 1) { println("Collected $value") }}} catch (e: Throwable) {println("Caught $e")}}

运行结果：

Emitting 1
1
Emitting 2
2
Collected 2
Caught java.lang.IllegalStateException: kotlin.Unit

构建端异常也可以使用 try-catch 在收集端捕获：

	private fun simple1(): Flow<String> =flow {for (i in 1..3) {println("Emitting $i")emit(i) // emit next value}}.map { value ->check(value <= 1) { "Crashed on $value" }"string $value"}@Testfun test11() = runBlocking<Unit> {try {simple1().collect { value -> println(value) }} catch (e: Throwable) {println("Caught $e")}}

运行结果：

Emitting 1
string 1
Emitting 2
Caught java.lang.IllegalStateException: Crashed on 2

但是课程中说不建议在收集端捕获发射端的异常，而是使用 catch 函数直接在发射端捕获：

	fun test12() = runBlocking<Unit> {flow {emit(1)throw ArithmeticException()}.catch { e: Throwable ->println("Caught $e")}.flowOn(Dispatchers.IO).collect { println(it) }}

使用 catch 函数捕获异常，运行结果：

Caught java.lang.ArithmeticException
1

此外还可以在 catch 中再次发射数据：

	fun test12() = runBlocking<Unit> {flow {emit(1)throw ArithmeticException()}.catch { e: Throwable ->println("Caught $e")emit(10)}.flowOn(Dispatchers.IO).collect { println(it) }}

运行结果：

1
Caught java.lang.ArithmeticException
10

3.2 流的完成

（P75）当流的收集完成（无论是正常完成还是异常完成）时，可能需要执行某些操作。可以通过两种方式来实现：命令式或声明式。

命令式是通过 finally 块在收集完成时执行操作：

	fun test13() = runBlocking<Unit> {try {(1..3).asFlow().collect { value -> println(value) }} finally {println("Done")}}

运行结果会在收集到所有数据后打印 Done：

1
2
3
Done

声明式则通过 onCompletion 中转操作符，它在流完全收集时调用：

	fun test14() = runBlocking<Unit> {(1..3).asFlow().onCompletion { println("Done") }.collect { value -> println(value) }}

运行结果：

1
2
3
Done

onCompletion 的主要优点是其 lambda 表达式的可空参数 Throwable 可以用于确定流收集是正常完成还是有异常发生：

	private fun simple3() = flow {emit(1)throw RuntimeException()}@Testfun test15() = runBlocking<Unit> {simple3().onCompletion { cause -> if (cause != null) println("Flow completed exceptionally") }.catch { throwable -> println("Caught $throwable") }.collect { value -> println(value) }}

运行结果：

1
Flow completed exceptionally
Caught java.lang.RuntimeException

注意 onCompletion 并不能处理异常，它只是输出信息用于确定收集是否正常完成，处理异常的任务要交给接下来的 catch。

onCompletion 还有一个特点就是能观察到所有异常，并且仅在上流成功完成（没有取消或失败）的情况下接收一个 null 异常：

	private fun simple4() = (1..3).asFlow()@Testfun test16() = runBlocking<Unit> {// 没有异常的simple4().onCompletion { cause -> println(cause) }.collect { value -> println(value) }// 下流有异常的simple4().onCompletion { cause -> println("Flow completed with $cause") }.collect { value ->check(value <= 1) { "Collected $value" }println(value)}}

运行结果：

1
2
3
null
1
Flow completed with java.lang.IllegalStateException: Collected 2Collected 2
java.lang.IllegalStateException: Collected 2
...