文章目录

一、流的上下文
- 1、上下文保存
- 2、流收集函数原型
- 3、流发射函数原型
- 4、代码示例 - 查看流发射和收集的协程
- 5、代码示例 - 不能在不同协程中执行相同流的发射和收集操作
二、修改流发射的协程上下文
- 1、Flow#flowOn 函数原型
- 2、代码示例

一、流的上下文

1、上下文保存

Flow 异步流收集元素的操作 , 一般是在协程上下文中进行的 , 如 : 在协程中调用 Flow#collect 函数 , 收集元素 ;

收集元素时的协程上下文 , 会传递给发射元素的流构建器 , 作为流构建器的上下文 ;

Flow 异步流在收集元素时 , 才调用流构建器中的代码 , 收集元素操作在协程中执行 , 流构建器也同样在相同的协程中运行 ;

流收集元素和发射元素在相同的协程上下文中的属性 , 称为上下文保存 ;

2、流收集函数原型

Flow#collect 函数原型如下 : Flow#collect 函数由 suspend 关键字修饰 , 该函数是 suspend 挂起函数 , 因此该函数必须在协程中调用 ;

public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value: T) -> Unit): Unit =collect(object : FlowCollector<T> {override suspend fun emit(value: T) = action(value)})

3、流发射函数原型

Flow 异步流的构建器函数 : 流构建器不是 suspend 挂起函数 , 可以在普通的线程中运行 , 不必在协程中运行 ;

flow 构建器 :

public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)

asFlow 构建器 :

@FlowPreview
public fun <T> (() -> T).asFlow(): Flow<T> = flow {emit(invoke())
}

flowOf 构建器 :

public fun <T> flowOf(vararg elements: T): Flow<T> = flow {for (element in elements) {emit(element)}
}

4、代码示例 - 查看流发射和收集的协程

代码示例 : 在流收集时和流构建时 , 分别打印线程名称 , 查看是在哪个线程中执行的 ;

package kim.hsl.coroutineimport android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlockingclass MainActivity : AppCompatActivity() {override fun onCreate(savedInstanceState: Bundle?) {super.onCreate(savedInstanceState)setContentView(R.layout.activity_main)// 携程中调用挂起函数返回一个 Flow 异步流runBlocking {println("流收集时的协程上下文 : ${Thread.currentThread().name}")// 调用 Flow#collect 函数, 可以获取在异步流中产生的元素flowFunction().collect {// 每隔 500ms 即可拿到一个 Int 元素// 并且该操作是异步操作, 不会阻塞调用线程println(it)}}}/*** 使用 flow 构建器 Flow 异步流* 在该异步流中, 异步地产生 Int 元素*/suspend fun flowFunction() = flow<Int> {println("流构建器的上下文 : ${Thread.currentThread().name}")for (i in 0..2) {// 挂起函数 挂起 500ms// 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令// 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令delay(500)// 每隔 500ms 产生一个元素// 通过调用 FlowCollector#emit 生成一个元素emit(i)}}
}

执行结果 : 最终执行时 , 流构建器和流收集都是在主线程中执行的 , 这是由 runBlocking 协程构建器将主线程包装后的协程 ;

2022-12-23 14:29:06.315 17484-17484/kim.hsl.coroutine I/System.out: 流收集时的协程上下文 : main
2022-12-23 14:29:06.323 17484-17484/kim.hsl.coroutine I/System.out: 流构建器的上下文 : main
2022-12-23 14:29:06.875 17484-17484/kim.hsl.coroutine I/System.out: 0
2022-12-23 14:29:07.399 17484-17484/kim.hsl.coroutine I/System.out: 1
2022-12-23 14:29:07.940 17484-17484/kim.hsl.coroutine I/System.out: 2

5、代码示例 - 不能在不同协程中执行相同流的发射和收集操作

在流构建器中 , 将代码定义在如下协程中执行 , 使用 Dispatchers.IO 调度器 , 也就是协程在子线程中执行 ;

withContext(Dispatchers.IO){}

在流收集时 , 在使用 runBlocking 将主线程包装后的协程中 , 收集元素 , 协程在主线程中执行 ;

runBlocking {}

代码示例 :

package kim.hsl.coroutineimport android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContextclass MainActivity : AppCompatActivity() {override fun onCreate(savedInstanceState: Bundle?) {super.onCreate(savedInstanceState)setContentView(R.layout.activity_main)// 携程中调用挂起函数返回一个 Flow 异步流runBlocking {println("流收集时的协程上下文 : ${Thread.currentThread().name}")// 调用 Flow#collect 函数, 可以获取在异步流中产生的元素flowFunction().collect {// 每隔 500ms 即可拿到一个 Int 元素// 并且该操作是异步操作, 不会阻塞调用线程println(it)}}}/*** 使用 flow 构建器 Flow 异步流* 在该异步流中, 异步地产生 Int 元素*/suspend fun flowFunction() = flow<Int> {// 在后台线程中发射元素withContext(Dispatchers.IO){println("流构建器的上下文 : ${Thread.currentThread().name}")for (i in 0..2) {// 挂起函数 挂起 500ms// 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令// 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令delay(500)// 每隔 500ms 产生一个元素// 通过调用 FlowCollector#emit 生成一个元素emit(i)}}}
}

执行程序后的报错信息如下 :

2022-12-23 14:39:05.805 19710-19710/kim.hsl.coroutine I/System.out: 流收集时的协程上下文 : main
2022-12-23 14:39:05.850 19710-19738/kim.hsl.coroutine I/System.out: 流构建器的上下文 : DefaultDispatcher-worker-1
2022-12-23 14:39:06.436 19710-19710/kim.hsl.coroutine D/AndroidRuntime: Shutting down VM
2022-12-23 14:39:06.462 19710-19710/kim.hsl.coroutine E/AndroidRuntime: FATAL EXCEPTION: mainProcess: kim.hsl.coroutine, PID: 19710java.lang.RuntimeException: Unable to start activity ComponentInfo{kim.hsl.coroutine/kim.hsl.coroutine.MainActivity}: java.lang.IllegalStateException: Flow invariant is violated:Flow was collected in [BlockingCoroutine{Active}@daf39f2, BlockingEventLoop@a6f9843],but emission happened in [DispatchedCoroutine{Active}@8ba6ec0, Dispatchers.IO].Please refer to 'flow' documentation or use 'flowOn' insteadat android.app.ActivityThread.performLaunchActivity(ActivityThread.java:2951)at android.app.ActivityThread.handleLaunchActivity(ActivityThread.java:3086)at android.app.servertransaction.LaunchActivityItem.execute(LaunchActivityItem.java:78)at android.app.servertransaction.TransactionExecutor.executeCallbacks(TransactionExecutor.java:108)at android.app.servertransaction.TransactionExecutor.execute(TransactionExecutor.java:68)at android.app.ActivityThread$H.handleMessage(ActivityThread.java:1816)at android.os.Handler.dispatchMessage(Handler.java:106)at android.os.Looper.loop(Looper.java:193)at android.app.ActivityThread.main(ActivityThread.java:6718)at java.lang.reflect.Method.invoke(Native Method)at com.android.internal.os.RuntimeInit$MethodAndArgsCaller.run(RuntimeInit.java:493)at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:858)Caused by: java.lang.IllegalStateException: Flow invariant is violated:Flow was collected in [BlockingCoroutine{Active}@daf39f2, BlockingEventLoop@a6f9843],but emission happened in [DispatchedCoroutine{Active}@8ba6ec0, Dispatchers.IO].Please refer to 'flow' documentation or use 'flowOn' insteadat 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)at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:59)at kim.hsl.coroutine.MainActivity$flowFunction$2$1.invokeSuspend(MainActivity.kt:43)at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:571)at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:750)at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:678)at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:665)
2022-12-23 14:39:06.505 19710-19710/kim.hsl.coroutine I/Process: Sending signal. PID: 19710 SIG: 9

二、修改流发射的协程上下文

在上述流的收集和流的发射都必须在同一个协程中执行 , 这样并不是我们想要的 ;

如 : 下载时 , 想要在后台线程中下载 , 在主线程中更新 UI , 那么对应 Flow 异步流应该是在后台线程中发射元素 , 在主线程中收集元素 ;

使用 flowOn 操作符 , 可以修改流发射的协程上下文 , 不必必须在流收集的协程上下文中执行流发射操作 ;

1、Flow#flowOn 函数原型

Flow#flowOn 函数原型如下 :

/*** 将此流执行的上下文更改为给定的[context]。* 此操作符是可组合的，仅影响前面没有自己上下文的操作符。* 这个操作符是上下文保护的:[context] **不会**泄漏到下游流中。** 例如:** ```* withContext(Dispatchers.Main) {*     val singleValue = intFlow // will be executed on IO if context wasn't specified before*         .map { ... } // Will be executed in IO*         .flowOn(Dispatchers.IO)*         .filter { ... } // Will be executed in Default*         .flowOn(Dispatchers.Default)*         .single() // Will be executed in the Main* }* ```** 有关上下文保存的更多说明，请参考[Flow]文档。** 如果更改上下文不需要更改，则此操作符保留流的_sequential_性质* (调度)[CoroutineDispatcher]。否则，如果需要更改dispatcher，它将进行收集* 使用指定[上下文]运行的协同例程中的流发射，并从另一个协同例程中发射它们* 使用带有[default][channel]的通道与原始收集器的上下文连接。BUFFERED]缓冲区大小* 在两个协程之间，类似于[buffer]操作符，除非显式调用[buffer]操作符* 在' flowOn '之前或之后，请求缓冲行为并指定通道大小。** 注意，跨不同调度程序操作的流在取消时可能会丢失一些正在运行的元素。* 特别是，该操作符确保下游流不会在取消时恢复，即使元素* 已经被上游的气流释放出来了。** ###算子融合** 相邻的[channelFlow]、[flowOn]、[buffer]和[produceIn]的应用是* 始终融合，以便只有一个正确配置的通道用于执行。** 多个“flowOn”操作符融合到一个具有组合上下文的单一“flowOn”。上下文的要素* 第一个' flowOn '操作符自然优先于第二个' flowOn '操作符的元素* 当它们具有相同的上下文键时，例如:** ```* flow.map { ... } // Will be executed in IO*     .flowOn(Dispatchers.IO) // This one takes precedence*     .flowOn(Dispatchers.Default)* ```** 请注意，[SharedFlow]的实例本身没有执行上下文，* 所以应用' flowOn '到' SharedFlow '没有效果。参见[SharedFlow]关于Operator Fusion的文档。** @throws [IllegalArgumentException] 如果所提供的上下文包含[Job]实例。*/
public fun <T> Flow<T>.flowOn(context: CoroutineContext): Flow<T> {checkFlowContext(context)return when {context == EmptyCoroutineContext -> thisthis is FusibleFlow -> fuse(context = context)else -> ChannelFlowOperatorImpl(this, context = context)}
}

2、代码示例