Kotlin（二十）异步流-操作符<2>
-
流上下文
流的收集总是在调用协程的上下文中发生。例如，如果有一个流 simple，然后以下代码在它的编写者指定的上下文中运行，而无论流 simple 的实现细节如何：
withContext(context) {
simple().collect { value ->
println(value) // 运行在指定上下文中
}
}
流的该属性称为 上下文保存 。
所以默认的，flow { ... } 构建器中的代码运行在相应流的收集器提供的上下文中。举例来说，考虑打印线程的一个 simple 函数的实现， 它被调用并发射三个数字：
fun simple(): Flow<Int> = flow {
log("Started simple flow")
for (i in 1..3) {
emit(i)
}
}
fun main() = runBlocking<Unit> {
simple().collect { value -> log("Collected $value") }
}
[main @coroutine#1] Started simple flow
[main @coroutine#1] Collected 1
[main @coroutine#1] Collected 2
[main @coroutine#1] Collected 3
由于 simple().collect 是在主线程调用的，那么 simple 的流主体也是在主线程调用的。 这是快速运行或异步代码的理想默认形式，它不关心执行的上下文并且不会阻塞调用者。
-
withContext 发出错误
然而，长时间运行的消耗 CPU 的代码也许需要在 Dispatchers.Default 上下文中执行，并且更新 UI 的代码也许需要在 Dispatchers.Main 中执行。通常，withContext 用于在 Kotlin 协程中改变代码的上下文，但是 flow {...} 构建器中的代码必须遵循上下文保存属性，并且不允许从其他上下文中发射（emit）。
报错代码
fun simple(): Flow<Int> = flow {
// 在流构建器中更改消耗 CPU 代码的上下文的错误方式
kotlinx.coroutines.withContext(Dispatchers.Default) {
for (i in 1..3) {
Thread.sleep(100) // 假装我们以消耗 CPU 的方式进行计算
emit(i) // 发射下一个值
}
}
}
fun main() = runBlocking<Unit> {
simple().collect { value -> println(value) }
}
error
Exception in thread "main" java.lang.IllegalStateException: Flow invariant is violated:
Flow was collected in [CoroutineId(1), "coroutine#1":BlockingCoroutine{Active}@4a03ee16, BlockingEventLoop@c4f7368],
but emission happened in [CoroutineId(1), "coroutine#1":DispatchedCoroutine{Active}@673ab05c, Dispatchers.Default].
Please refer to 'flow' documentation or use 'flowOn' instead
at kotlinx.coroutines.flow.internal.SafeCollector_commonKt.checkContext (SafeCollector.common.kt:84)
at kotlinx.coroutines.flow.internal.SafeCollector.checkContext (SafeCollector.kt:88)
at kotlinx.coroutines.flow.internal.SafeCollector.emit (SafeCollector.kt:74)
-
flowOn 操作符
例外的是 flowOn 函数，该函数用于更改流发射的上下文。 以下示例展示了更改流上下文的正确方法，该示例还通过打印相应线程的名字以展示它们的工作方式：
fun simple(): Flow<Int> = flow {
for (i in 1..3) {
Thread.sleep(100) // 假装我们以消耗 CPU 的方式进行计算
log("Emitting $i")
emit(i) // 发射下一个值
}
}.flowOn(Dispatchers.Default) // 在流构建器中改变消耗 CPU 代码上下文的正确方式
fun main() = runBlocking<Unit> {
simple().collect { value ->
log("Collected $value")
}
}
[DefaultDispatcher-worker-1 @coroutine#2] Emitting 1
[main @coroutine#1] Collected 1
[DefaultDispatcher-worker-1 @coroutine#2] Emitting 2
[main @coroutine#1] Collected 2
[DefaultDispatcher-worker-1 @coroutine#2] Emitting 3
[main @coroutine#1] Collected 3
这里要观察的另一件事是 flowOn 操作符已改变流的默认顺序性。 现在收集发生在一个协程中（“coroutine#1”）而发射发生在运行于另一个线程中与收集协程并发运行的另一个协程（“coroutine#2”）中。当上游流必须改变其上下文中的 CoroutineDispatcher 的时候，flowOn 操作符创建了另一个协程。
-
缓冲
从收集流所花费的时间来看，将流的不同部分运行在不同的协程中将会很有帮助，特别是当涉及到长时间运行的异步操作时。例如，考虑一种情况， 一个 simple 流的发射很慢，它每花费 100 毫秒才产生一个元素；而收集器也非常慢， 需要花费 300 毫秒来处理元素。让我们看看从该流收集三个数字要花费多长时间：
fun simple(): Flow<Int> = flow {
for (i in 1..3) {
delay(100) // 假装我们异步等待了 100 毫秒
emit(i) // 发射下一个值
}
}
fun main() = runBlocking<Unit> {
val time = measureTimeMillis {
simple().collect { value ->
delay(300) // 假装我们花费 300 毫秒来处理它
println(value)
}
}
println("Collected in $time ms")
}
1
2
3
Collected in 1217 ms
它会产生这样的结果，整个收集过程大约需要 1200 毫秒（3 个数字，每个花费 400 毫秒）：
我们可以在流上使用 buffer 操作符来并发运行这个 simple 流中发射元素的代码以及收集的代码， 而不是顺序运行它们：
val time = measureTimeMillis {
simple()
.buffer() // 缓冲发射项，无需等待
.collect { value ->
delay(300) // 假装我们花费 300 毫秒来处理它
println(value)
}
}
println("Collected in $time ms")
1
2
3
Collected in 1047 ms
它产生了相同的数字，只是更快了，由于我们高效地创建了处理流水线， 仅仅需要等待第一个数字产生的 100 毫秒以及处理每个数字各需花费的 300 毫秒。这种方式大约花费了 1000 毫秒来运行：
注意，当必须更改 时，[flowOn]操作符使用了相同的缓冲机制， 但是我们在这里显式地请求缓冲而不改变执行上下文。
-
合并
当流代表部分操作结果或操作状态更新时，可能没有必要处理每个值，而是只处理最新的那个。在本示例中，当收集器处理它们太慢的时候， conflate 操作符可以用于跳过中间值。构建前面的示例：
val time = measureTimeMillis {
simple()
.conflate() // 合并发射项，不对每个值进行处理
.collect { value ->
delay(300) // 假装我们花费 300 毫秒来处理它
println(value)
}
}
println("Collected in $time ms")
1
3
Collected in 746 ms
我们看到，虽然第一个数字仍在处理中，但第二个和第三个数字已经产生，因此第二个是 conflated ，只有最新的（第三个）被交付给收集器
