协程: Flow 异步流 /

以异步方式返回多个返回值的方案:

在 Kotlin 协程 Coroutine 中 , 使用 suspend 挂起函数以异步的方式返回单个返回值肯定可以实现 , 如果要以异步的方式返回多个元素的返回值 , 可以使用如下方案 :

集合
序列
Suspend 挂起函数
Flow 异步流

同步调用返回多个值的弊端：

同步调用函数时 , 如果函数耗时太长或者中途有休眠 , 则会阻塞主线程导致 ANR 异常 ;

协程中调用挂起函数返回集合

如果要以异步方式返回多个返回值 , 可以在协程中调用挂起函数返回集合 , 但是该方案只能一次性返回多个返回值 , 不能持续不断的先后返回多个返回值 ;

// 携程中调用挂起函数返回多个值// 调用 " 返回 List 集合的挂起函数 " , 并遍历返回值runBlocking {listFunction().forEach {// 遍历打印集合中的内容Log.e(TAG, "$it")}}

使用 Flow 异步流持续获取不同返回值

序列可以先后返回多个返回值 , 但是会阻塞线程 ;序列可以先后返回多个返回值 , 但是会阻塞线程 ;

本篇博客中开始引入 Flow 异步流的方式 , 持续性返回多个返回值 ;

调用 flow 构建器 , 可创建 Flow 异步流 , 在该异步流中, 异步地产生指定类型的元素 ;

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

在 flow 异步流构建器中 , 通过调用 FlowCollector#emit 生成一个元素 ; 函数原型如下 :

/*** [FlowCollector]用作流的中间或终端收集器，并表示接收[Flow]发出的值的实体。* 该接口通常不应该直接实现，而是在实现自定义操作符时作为[flow]构建器中的接收器使用。* 这个接口的实现不是线程安全的。*/
public interface FlowCollector<in T> {/*** 收集上游发出的值。* 此方法不是线程安全的，不应该并发调用。*/public suspend fun emit(value: T)
}

调用 Flow#collect 函数, 可以获取在异步流中产生的元素 , 并且该操作是异步操作, 不会阻塞调用线程 ;

public interface Flow<out T> {/*** 接收给定的[collector]并[发出][FlowCollector]。向它发射]值。* 永远不应该实现或直接使用此方法。** 直接实现“Flow”接口的唯一方法是扩展[AbstractFlow]。* 要将它收集到特定的收集器，可以使用' collector. emitall (flow) '或' collect{…}的扩展* 应该使用。这样的限制确保了上下文保存属性不被侵犯，并防止了大多数情况* 与并发性、不一致的流调度程序和取消相关的开发人员错误。*/@InternalCoroutinesApipublic suspend fun collect(collector: FlowCollector<T>)
}

案例

class FlowActivity  : ComponentActivity() {override   fun onCreate(savedInstanceState: Bundle?){// 协程中调用挂起函数flowFunction()返回一个 Flow 异步流runBlocking{// 调用 Flow#collect 函数, 可以获取在Flow异步流中产生的元素值itval mFlow: Flow<Int> = flowFunction()mFlow.collect(collector = {// 每隔 500ms 即可  获取    Flow异步流中的一个Int 元素// 并且该操作是异步操作, 不会阻塞调用线程Log.e(TAG, "获取Flow异步流中的一个Int元素it=$it")})}
/*** 获取Flow异步流中的一个Int元素it=0* 获取Flow异步流中的一个Int元素it=1* 获取Flow异步流中的一个Int元素it=2* */}/*** 使用 flow 构建器 Flow 异步流* 在该异步流中, 异步地产生 Int 元素*/suspend fun flowFunction(): Flow<Int>{val flow : Flow<Int> = flow<Int>(block = {Log.e(TAG, "输出接受者对象this=${this}")for (i in 0..2) {// 挂起函数 挂起 500ms// 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令// 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令delay(500)// 每隔 500ms 产生一个元素// 通过调用 FlowCollector#emit 生成一个元素this.emit(i)}})return  flow}}

Flow 异步流获取返回值方式与其它方式对比

① 异步流构建方式 : Flow 异步流是通过 flow 构建器函数创建的 ;

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

② 构建器可调用挂起函数 : flow 构建器代码块中的代码 , 是可以挂起的 , 可以在其中调用挂起函数 , 如 kotlinx.coroutines.delay 函数等 ;

 /*** 使用 flow 构建器 Flow 异步流* 在该异步流中, 异步地产生 Int 元素*/suspend fun flowFunction(): Flow<Int>{val flow : Flow<Int> = flow<Int>(block = {Log.e(TAG, "输出接受者对象this=${this}")for (i in 0..2) {// 挂起函数 挂起 500ms// 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令// 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令delay(500)// 每隔 500ms 产生一个元素// 通过调用 FlowCollector#emit 生成一个元素this.emit(i)}})return  flow}

③ suspend 关键字可省略 : 返回值为 Flow 异步流的函数 , 其默认就是 suspend 挂起函数 , suspend 关键字可以省略 , 上述函数中不标注 suspend 也可 ;

 /*** 使用 flow 构建器 Flow 异步流* 在该异步流中, 异步地产生 Int 元素*///suspend fun flowFunction(): Flow<Int>{fun flowFunction(): Flow<Int>{val flow : Flow<Int> = flow<Int>(block = {Log.e(TAG, "输出接受者对象this=${this}")for (i in 0..2) {// 挂起函数 挂起 500ms// 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令// 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令delay(500)// 每隔 500ms 产生一个元素// 通过调用 FlowCollector#emit 生成一个元素this.emit(i)}})return  flow}

④ 生成元素 : 在 Flow 异步流中 , 通过调用 FlowCollector#emit 函数生成元素 ;

⑤ 收集元素 : 在 Flow 异步流中 , 通过调用 Flow#collect 函数可以收集在 Flow 异步流中生成的元素 ;

在 Android 中使用 Flow 异步流下载文件

Android 中主线程不可执行网络相关操作 , 因此只能在子线程中下载文件 ,可以在协程中使用 Dispatcher.IO 调度器在子线程下载文件 ,下载文件时需要实时显示下载百分比进度 ,这个进度需要上报给主线程 , 在主线程中更新 UI 显示下载进度 ,在 Flow 异步流中 , 可以使FlowCollector#emit 向主线程中发送进度值。在主线程中 , 可以使用 Flow#collect 函数收集 Flow 异步流中发射出来的数据 , 如 : 进度 , 捕获的异常 , 下载状态等 ;

完整流程 , 如下图所示 :

Flow 冷流 ( 流被收集时运行 )

Flow 异步流的构建器函数 flow 函数中的代码 ,在调用 Flow#collect 函数时 , 也就是在 Flow 异步流收集元素时 ,才会执行 flow 构建器中的代码 ;这种机制的异步流称为冷流 ;

Flow 异步流冷流代码示例 :

在 flow 构建器的开始位置 , 发射元素 , 在主线程中 Flow#collect 收集元素位置 , 添加日志信息 , 查看日志打印的时机 ;

class FlowActivity  : ComponentActivity() {override   fun onCreate(savedInstanceState: Bundle?){// 协程中调用挂起函数flowFunction()返回一个 Flow 异步流runBlocking{// 调用 Flow#collect 函数, 可以获取在Flow异步流中产生的元素值itval mFlow: Flow<Int> = flowFunction()Log.e(TAG, "Flow 异步流冷流 开始收集元素")mFlow.collect(collector = {// 每隔 500ms 即可  获取    Flow异步流中的一个Int 元素// 并且该操作是异步操作, 不会阻塞调用线程Log.e(TAG, "收集Flow异步流冷流中的一个个元素it=$it")})}
/**
Flow 异步流冷流 开始收集元素
Flow 异步流冷流开始发射元素Flow 异步流冷流发射元素值i=0
收集Flow异步流冷流中的一个个元素it=0Flow 异步流冷流发射元素值i=1
收集Flow异步流冷流中的一个个元素it=1Flow 异步流冷流发射元素值i=2
收集Flow异步流冷流中的一个个元素it=2
*/}/*** 使用 flow 构建器 Flow 异步流* 在该异步流中, 异步地产生 Int 元素*/suspend fun flowFunction(): Flow<Int>{val flow : Flow<Int> = flow<Int>(block = {Log.e(TAG, "输出接受者对象this=${this}")Log.e(TAG,"Flow 异步流冷流开始发射元素")for (i in 0..2) {// 挂起函数 挂起 500ms// 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令// 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令delay(500)// 每隔 500ms 产生一个元素// 通过调用 FlowCollector#emit 生成一个元素Log.e(TAG,"Flow 异步流冷流发射元素值i=$i")this.emit(i)}})return  flow}}

Flow 流的连续性

Flow 流的每次调用 Flow#collect 收集元素的操作 , 都是按照固定顺序执行的 , 使用特殊操作符可以改变该顺序 ;

Flow 异步流中的元素 , 按照顺序进行 FlowCollector#emit 发射元素操作 , 则调用 Flow#collect 收集元素时获取的元素也是按照顺序获取的 ;

在流的上游发射元素到下游收集元素的过程中 , 会使用过渡操作符处理每个 FlowCollector#emit 发射出的元素 , 最终才交给最末端的操作符 ;

class FlowActivity  : ComponentActivity() {override   fun onCreate(savedInstanceState: Bundle?){// 协程中调用挂起函数返回一个 Flow 异步流runBlocking{// 使用下面的方式asFlow()可以快速构建一个 Flow 流//上游发射元素val mFlow : Flow<Int> = (0..5).asFlow()val filterFlow: Flow<Int> = mFlow.filter (predicate={// Log.e(TAG, "下游收集元素it=$it")// 奇数才能继续向下流,偶数被过滤掉了it % 2 == 1})filterFlow.collect(collector = {Log.e(TAG, "下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=$it")})//将Flow异步流冷流: filterFlow的元素值进行转换后，返回一个新的Flow流 ,然后对流的元素进行发射val mapFlow : Flow<String> = filterFlow.map<Int,String>(transform={// 遍历元素, 将其拼接成字符串"学号 : $it"})mapFlow.collect(collector = {Log.e(TAG, "下游收集Flow异步流冷流mapFlow的一个个元素it=$it")})}/** 下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=1*  下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=3*  下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=5*  下游收集Flow异步流冷流mapFlow的一个个元素it=学号 : 1* 下游收集Flow异步流冷流mapFlow的一个个元素it=学号 : 3* 下游收集Flow异步流冷流mapFlow的一个个元素it=学号 : 5* */}}