• 协程: Flow 异步流 /

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

     案例

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.       // 协程中调用挂起函数flowFunction()返回一个 Flow 异步流
    6.       runBlocking{
    7.                // 调用 Flow#collect 函数, 可以获取在Flow异步流中产生的元素值it
    8.                val mFlow: Flow<Int> = flowFunction()
    9.                mFlow.collect(collector = {
    10.                    Log.e(TAG," 收集Flow异步流冷流mFlow的协程上下文 : ${Thread.currentThread().name}")
    11.                 // 每隔 500ms 即可  获取    Flow异步流中的一个Int 元素
    12.                    // 并且该操作是异步操作, 不会阻塞调用线程
    13.                    Log.e(TAG, "收集Flow异步流冷流mFlow中的一个个元素it=$it")
    14.                })
    15.            }
    16. /**
    17. Flow 异步流冷流mFlow的构建器的上下文 : main
    18. 收集Flow异步流冷流mFlow的协程上下文 : main
    20. Flow 异步流冷流mFlow发射元素值i=0
    21. 收集Flow异步流冷流mFlow中的一个个元素it=0
    23. Flow 异步流冷流mFlow发射元素值i=1
    24. 收集Flow异步流冷流mFlow中的一个个元素it=1
    26. Flow 异步流冷流mFlow发射元素值i=2
    27. 收集Flow异步流冷流mFlow中的一个个元素it=2
    29. */
    30.  
    31.  
    32. }
    33.  
    34.  
    35.    /**
    36.      * 使用 flow 构建器 Flow 异步流
    37.      * 在该异步流中, 异步地产生 Int 元素
    38.      */
    39.  
    40.    suspend fun flowFunction(): Flow<Int>{
    41.        val mFlow : Flow<Int> = flow<Int>(block = {
    42.             Log.e(TAG, "输出接受者对象this=${this}")
    43.            Log.e(TAG, "Flow 异步流冷流mFlow的构建器的上下文 : ${Thread.currentThread().name}")
    44.             for (i in 0..2) {
    45.                 // 挂起函数 挂起 500ms
    46.                 // 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令
    47.                 // 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令
    48.                 delay(500)
    49.                 // 每隔 500ms 产生一个元素
    50.                 // 通过调用 FlowCollector#emit 生成一个元素
    51.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i")
    52.                 this.emit(i)
    53.             }
    54.         })
    55.         return  mFlow
    56.     }
    57.  
    58. }

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

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

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

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

    1.    /**
    2.      * 使用 flow 构建器 Flow 异步流
    3.      * 在该异步流中, 异步地产生 Int 元素
    4.      */
    5.  
    6.    suspend fun flowFunction(): Flow<Int>{
    7.        val mFlow : Flow<Int> = flow<Int>(block = {
    8.             Log.e(TAG, "输出接受者对象this=${this}")
    9.            Log.e(TAG, "Flow 异步流冷流mFlow的构建器的上下文 : ${Thread.currentThread().name}")
    10.             for (i in 0..2) {
    11.                 // 挂起函数 挂起 500ms
    12.                 // 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令
    13.                 // 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令
    14.                 delay(500)
    15.                 // 每隔 500ms 产生一个元素
    16.                 // 通过调用 FlowCollector#emit 生成一个元素
    17.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i")
    18.                 this.emit(i)
    19.             }
    20.         })
    21.         return  mFlow
    22.     }

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

    1.   /**
    2.      * 使用 flow 构建器 Flow 异步流
    3.      * 在该异步流中, 异步地产生 Int 元素
    4.      */
    5.  
    6.  //  suspend fun flowFunction(): Flow{
    7.   fun flowFunction(): Flow<Int>{
    8.        val mFlow : Flow<Int> = flow<Int>(block = {
    9.             Log.e(TAG, "输出接受者对象this=${this}")
    10.            Log.e(TAG, "Flow 异步流冷流mFlow的构建器的上下文 : ${Thread.currentThread().name}")
    11.             for (i in 0..2) {
    12.                 // 挂起函数 挂起 500ms
    13.                 // 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令
    14.                 // 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令
    15.                 delay(500)
    16.                 // 每隔 500ms 产生一个元素
    17.                 // 通过调用 FlowCollector#emit 生成一个元素
    18.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i")
    19.                 this.emit(i)
    20.             }
    21.         })
    22.         return  mFlow
    23.     }
    24.

    ④ 生成元素 : 在 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 收集元素位置 , 添加日志信息 , 查看日志打印的时机 ;

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.    // 协程中调用挂起函数flowFunction()返回一个 Flow 异步流
    6.       runBlocking{
    7.                // 调用 Flow#collect 函数, 可以获取在Flow异步流中产生的元素值it
    8.                val mFlow: Flow<Int> = flowFunction()
    9.                mFlow.collect(collector = {
    10.                    Log.e(TAG," 收集Flow异步流冷流mFlow的协程上下文 : ${Thread.currentThread().name}")
    11.                 // 每隔 500ms 即可  获取    Flow异步流中的一个Int 元素
    12.                    // 并且该操作是异步操作, 不会阻塞调用线程
    13.                    Log.e(TAG, "收集Flow异步流冷流mFlow中的一个个元素it=$it")
    14.                })
    15.            }
    16. /**
    17. Flow 异步流冷流mFlow的构建器的上下文 : main
    18. 收集Flow异步流冷流mFlow的协程上下文 : main
    20. Flow 异步流冷流mFlow发射元素值i=0
    21. 收集Flow异步流冷流mFlow中的一个个元素it=0
    23. Flow 异步流冷流mFlow发射元素值i=1
    24. 收集Flow异步流冷流mFlow中的一个个元素it=1
    26. Flow 异步流冷流mFlow发射元素值i=2
    27. 收集Flow异步流冷流mFlow中的一个个元素it=2
    29. */
    30.  
    31.  
    32. }
    33.  
    34.   
    35.    /**
    36.      * 使用 flow 构建器 Flow 异步流
    37.      * 在该异步流中, 异步地产生 Int 元素
    38.      */
    39.  
    40.    suspend fun flowFunction(): Flow<Int>{
    41.        val mFlow : Flow<Int> = flow<Int>(block = {
    42.             Log.e(TAG, "输出接受者对象this=${this}")
    43.            Log.e(TAG, "Flow 异步流冷流mFlow的构建器的上下文 : ${Thread.currentThread().name}")
    44.             for (i in 0..2) {
    45.                 // 挂起函数 挂起 500ms
    46.                 // 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令
    47.                 // 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令
    48.                 delay(500)
    49.                 // 每隔 500ms 产生一个元素
    50.                 // 通过调用 FlowCollector#emit 生成一个元素
    51.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i")
    52.                 this.emit(i)
    53.             }
    54.         })
    55.         return  mFlow
    56.     }
    57.  
    58. }

    Flow 流的连续性

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

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

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

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.     // 协程中调用挂起函数返回一个 Flow 异步流
    6.  
    7.         runBlocking{
    8.             // 使用下面的方式asFlow()可以快速构建一个 Flow 流
    9.               //上游发射元素
    10.             val mFlow : Flow<Int> = (0..5).asFlow()
    11.            val filterFlow: Flow<Int> = mFlow.filter (predicate={
    12.                  if( it % 2 == 1){
    13.                      Log.e(TAG,"Flow 异步流冷流filterFlow发射过滤后符合条件的元素值it=${it}")
    14.                  }
    15.                // 奇数才能继续向下流,偶数被过滤掉了
    16.                 it % 2 == 1 //Lambda最后一行最为返回值
    17.                 })
    18.             filterFlow.collect(collector = {
    19.               Log.e(TAG, "下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=$it")
    20.             })
    21.             //将Flow异步流冷流filterFlow的元素值进行转换后，返回一个新的Flow流 ,然后对流的元素进行发射
    22.            val mapFlow : Flow = filterFlow.map<Int,String>(transform={
    23.                 // 遍历元素, 将其拼接成字符串
    24.                val str = "学号 : $it"
    25.                Log.e(TAG,"Flow 异步流冷流mapFlow发射转换后的元素值str=$str")
    26.                str //Lambda最后一行最为返回值
    27.             })
    28.             mapFlow.collect(collector = {
    29.                 Log.e(TAG, "下游收集Flow异步流冷流mapFlow的一个个元素it=$it")
    30.             })
    31.  
    32.         }
    33.  
    34.  
    35.    /**
    36.     * Flow 异步流冷流filterFlow发射过滤后符合条件的元素值it=1
    37.     *下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=1
    38.     * Flow 异步流冷流filterFlow发射过滤后符合条件的元素值it=3
    39.     *  下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=3
    40.     *  Flow 异步流冷流filterFlow发射过滤后符合条件的元素值it=5
    41.     *   下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=5
    42.     *
    43.     *  Flow 异步流冷流mapFlow发射转换后的元素值str=学号 : 1
    44.     *  下游收集Flow异步流冷流mapFlow的一个个元素it=学号 : 1
    45.     *  Flow 异步流冷流mapFlow发射转换后的元素值str=学号 : 3
    46.     *  下游收集Flow异步流冷流mapFlow的一个个元素it=学号 : 3
    47.     *  Flow 异步流冷流mapFlow发射转换后的元素值str=学号 : 5
    48.     *  下游收集Flow异步流冷流mapFlow的一个个元素it=学号 : 5
    49.     *
    50.     * */ 
    51.  
    52. }
    53.  
    54. }

    Flow流的构建器函数

    1、使用 flow()函数 构建 Flow异步流冷流

     在 flow 流构建器中 , 调用 FlowCollector#emit 函数 发射元素 , 然后在外部 调用 Flow#collect 函数 收集元素 ;

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

    2、使用flowOf()函数 构建 Flow异步流冷流

    1. public fun  flowOf(vararg elements: T): Flow = flow {
    2.     for (element in elements) {
    3.         emit(element)
    4.     }
    5. }
    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  runBlocking {
    5.        // 协程中调用挂起函数返回一个 Flow 异步流
    6.        val mFlowOf: Flow<Int> = flowOf(0, 1, 2, 3)
    7.        val mOnEachFlow: Flow<Int> =  mFlowOf.onEach {
    8.            // 每次发射元素时调用的代码块
    9.            delay(1000)
    10.            Log.e(TAG,"Flow 异步流冷流mOnEachFlow发射元素值it=$it")
    11.        }
    12.  
    13.        mOnEachFlow.collect(collector = {
    14.            // 每隔 1 秒接收一个元素
    15.            Log.e(TAG, "下游收集Flow异步流冷流mOnEachFlow的一个个元素it=$it")
    16.        })
    17.    }
    18.  
    19.   /**
    20.    *  Flow 异步流冷流mOnEachFlow发射元素值it=0
    21.    *下游收集Flow异步流冷流mOnEachFlow的一个个元素it=0
    22.    *
    23.    *  Flow 异步流冷流mOnEachFlow发射元素值it=1
    24.    * 下游收集Flow异步流冷流mOnEachFlow的一个个元素it=1
    25.    *
    26.    * Flow 异步流冷流mOnEachFlow发射元素值it=2
    27.    * 下游收集Flow异步流冷流mOnEachFlow的一个个元素it=2
    28.    *
    29.    *  Flow 异步流冷流mOnEachFlow发射元素值it=3
    30.    *  下游收集Flow异步流冷流mOnEachFlow的一个个元素it=3
    31.    * */
    32.  
    33. }
    34.  
    35. }

    3、使用asFlow()函数 构建 Flow异步流冷流

    1. public fun IntRange.asFlow(): Flow<Int> = flow {
    2.     forEach { value ->
    3.         emit(value)
    4.     }}
    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.   // 协程中调用挂起函数返回一个 Flow 异步流
    6.  
    7.         runBlocking{
    8.             // 使用下面的方式asFlow()可以快速构建一个 Flow 流
    9.               //上游发射元素
    10.             val mFlow : Flow<Int> = (0..5).asFlow()
    11.            val filterFlow: Flow<Int> = mFlow.filter (predicate={
    12.                  if( it % 2 == 1){
    13.                      Log.e(TAG,"Flow 异步流冷流filterFlow发射过滤后符合条件的元素值it=${it}")
    14.                  }
    15.                // 奇数才能继续向下流,偶数被过滤掉了
    16.                 it % 2 == 1 //Lambda最后一行最为返回值
    17.                 })
    18.             filterFlow.collect(collector = {
    19.               Log.e(TAG, "下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=$it")
    20.             })
    21.             //将Flow异步流冷流filterFlow的元素值进行转换后，返回一个新的Flow流 ,然后对流的元素进行发射
    22.            val mapFlow : Flow = filterFlow.map<Int,String>(transform={
    23.                 // 遍历元素, 将其拼接成字符串
    24.                val str = "学号 : $it"
    25.                Log.e(TAG,"Flow 异步流冷流mapFlow发射转换后的元素值str=$str")
    26.                str //Lambda最后一行最为返回值
    27.             })
    28.             mapFlow.collect(collector = {
    29.                 Log.e(TAG, "下游收集Flow异步流冷流mapFlow的一个个元素it=$it")
    30.             })
    31.  
    32.         }
    33.  
    34.  
    35.    /**
    36.     * Flow 异步流冷流filterFlow发射过滤后符合条件的元素值it=1
    37.     *下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=1
    38.     * Flow 异步流冷流filterFlow发射过滤后符合条件的元素值it=3
    39.     *  下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=3
    40.     *  Flow 异步流冷流filterFlow发射过滤后符合条件的元素值it=5
    41.     *   下游收集Flow异步流冷流filterFlow过滤后满足条件的一个个元素it=5
    42.     *
    43.     *  Flow 异步流冷流mapFlow发射转换后的元素值str=学号 : 1
    44.     *  下游收集Flow异步流冷流mapFlow的一个个元素it=学号 : 1
    45.     *  Flow 异步流冷流mapFlow发射转换后的元素值str=学号 : 3
    46.     *  下游收集Flow异步流冷流mapFlow的一个个元素it=学号 : 3
    47.     *  Flow 异步流冷流mapFlow发射转换后的元素值str=学号 : 5
    48.     *  下游收集Flow异步流冷流mapFlow的一个个元素it=学号 : 5
    49.     *
    50.     * */
    51.  
    52. }
    53.  
    54. }

    Flow 异步流的上下文

    1、上下文保存

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

    收集元素 时 的 协程上下文 , 会 传递给 发射元素 的 流构建器 , 作为 流构建器的 上下文 ;Flow 异步流 在 收集元素 时 , 才调用 流构建器 中的代码 , 收集元素操作在协程中执行 , 流构建器 也同样在相同的协程中运行 ;

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

    2、流收集函数原型

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

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

    3、流发射函数原型

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

    使用 flow()函数 构建 Flow异步流冷流

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

    使用flowOf()函数 构建 Flow异步流冷流

    1. public fun  flowOf(vararg elements: T): Flow = flow {
    2.     for (element in elements) {
    3.         emit(element)
    4.     }
    5. }

    使用asFlow()函数 构建 Flow异步流冷流

    1. @FlowPreview
    2. public fun  (() -> T).asFlow(): Flow = flow {
    3.     emit(invoke())
    4. }

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

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

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

    1.  
    2. class FlowActivity  : ComponentActivity() {
    3.  
    4.  override   fun onCreate(savedInstanceState: Bundle?){
    5.  
    6.   // 协程中调用挂起函数flowFunction()返回一个 Flow 异步流
    7.       runBlocking{
    8.                // 调用 Flow#collect 函数, 可以获取在Flow异步流中产生的元素值it
    9.                val mFlow: Flow<Int> = flowFunction()
    10.                mFlow.collect(collector = {
    11.                    Log.e(TAG," 收集Flow异步流冷流mFlow的协程上下文 : ${Thread.currentThread().name}")
    12.                 // 每隔 500ms 即可  获取    Flow异步流中的一个Int 元素
    13.                    // 并且该操作是异步操作, 不会阻塞调用线程
    14.                    Log.e(TAG, "收集Flow异步流冷流mFlow中的一个个元素it=$it")
    15.                })
    16.            }
    17. /**
    18. Flow 异步流冷流mFlow的构建器的上下文 : main
    19. 收集Flow异步流冷流mFlow的协程上下文 : main
    21. Flow 异步流冷流mFlow发射元素值i=0
    22. 收集Flow异步流冷流mFlow中的一个个元素it=0
    24. Flow 异步流冷流mFlow发射元素值i=1
    25. 收集Flow异步流冷流mFlow中的一个个元素it=1
    27. Flow 异步流冷流mFlow发射元素值i=2
    28. 收集Flow异步流冷流mFlow中的一个个元素it=2
    30. */
    31.  
    32. }
    33.  
    34.  
    35.  
    36.  /**
    37.      * 使用 flow 构建器 Flow 异步流
    38.      * 在该异步流中, 异步地产生 Int 元素
    39.      */
    40.  
    41.    suspend fun flowFunction(): Flow<Int>{
    42.        val mFlow : Flow<Int> = flow<Int>(block = {
    43.             Log.e(TAG, "输出接受者对象this=${this}")
    44.            Log.e(TAG, "Flow 异步流冷流mFlow的构建器的上下文 : ${Thread.currentThread().name}")
    45.             for (i in 0..2) {
    46.                 // 挂起函数 挂起 500ms
    47.                 // 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令
    48.                 // 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令
    49.                 delay(500)
    50.                 // 每隔 500ms 产生一个元素
    51.                 // 通过调用 FlowCollector#emit 生成一个元素
    52.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i")
    53.                 this.emit(i)
    54.             }
    55.         })
    56.         return  mFlow
    57.     }
    58.  
    59.  
    60. }

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

    修改流发射的协程上下文

    在上述 流的收集 和 流的发射 都 必须在同一个协程中执行 , 这样并不是我们想要的 ;如 : 下载时 , 想要在后台线程中下载 , 在主线程中更新 UI , 那么对应 Flow 异步流应该是在 后台线程中 发射元素 , 在主线程中 收集元素 ;

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

    1、Flow#flowOn 函数原型

    Flow#flowOn 函数原型如下 :

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

    2、代码示例

    流发射 在子线程中执行 , 流收集 在 主线程中执行 ;

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.  runBlocking{
    6.             // 调用 Flow#collect 函数, 可以获取在Flow异步流中产生的元素值it
    7.             val mFlow: Flow<Int> = flowFunction2()
    8.             mFlow.collect(collector = {
    9.                 // 每隔 500ms 即可  获取    Flow异步流中的一个Int 元素
    10.                 // 并且该操作是异步操作, 不会阻塞调用线程
    11.                 Log.e(TAG, "收集Flow异步流冷流mFlow中的一个个元素it=$it -------- mFlow的协程所在的上下文 : ${Thread.currentThread().name}")
    12.             })
    13.         }
    14.  
    15. /**
    16. Flow 异步流冷流mFlow发射元素值i=0 -------- mFlow构建器所在的上下文 : DefaultDispatcher-worker-3
    17. 收集Flow异步流冷流mFlow中的一个个元素it=0 -------- mFlow的协程所在的上下文 : main
    19. Flow 异步流冷流mFlow发射元素值i=1 -------- mFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    20. 收集Flow异步流冷流mFlow中的一个个元素it=1 -------- mFlow的协程所在的上下文 : main
    22. Flow 异步流冷流mFlow发射元素值i=2 -------- mFlow构建器所在的上下文 : DefaultDispatcher-worker-3
    23. 收集Flow异步流冷流mFlow中的一个个元素it=2 -------- mFlow的协程所在的上下文 : main
    25.  * */
    26.  
    27. }
    28.  
    29.   /**
    30.      * 使用 flow 构建器 Flow 异步流
    31.      * 在该异步流中, 异步地产生 Int 元素
    32.      */
    33.     suspend fun flowFunction2(): Flow<Int>{
    34.         val mFlow : Flow<Int> = flow<Int>(block = {
    35.             Log.e(TAG, "输出接受者对象this=${this}")
    36.  
    37.             for (i in 0..2) {
    38.                 // 挂起函数 挂起 500ms
    39.                 // 在协程中, 该挂起操作不会阻塞调用线程, 会继续执行其它代码指令
    40.                 // 500ms 恢复执行, 继续执行挂起函数之后的后续代码指令
    41.                 delay(500)
    42.                 // 每隔 500ms 产生一个元素
    43.                 // 通过调用 FlowCollector#emit 生成一个元素
    44.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i -------- mFlow构建器所在的上下文 : ${Thread.currentThread().name}")
    45.                 this.emit(i)
    46.             }
    47.         }).flowOn(context=Dispatchers.IO)
    48.         return  mFlow
    49.     }
    50.  
    51.  
    52. }

    调用 Flow#launchIn() 函数指定流收集的协程

    1、指定流收集的协程

    响应式编程 , 是 基于事件驱动 的 , 在 Flow 流中会产生源源不断的事件 , 就是 发射元素操作 ;拿到 Flow 流后 , 开始 收集元素 , 按照顺序逐个处理产生的事件 ( 元素 ) ;

    调用 Flow#launchIn()函数 , 传入 协程作用域 作为参数 , 可以 指定 收集 Flow 流元素 的 协程 ;

    我们知道调用 Flow#flowOn() 函数 , 可以 指定 Flow 流发射元素 的 协程 ;

    Flow#launchIn() 函数返回值是 Job 对象 , 是 协程任务对象 , 可调用 Job#cancel 函数取消该协程任务 ;

    2、Flow#launchIn() 函数原型

    1. /**
    2.  * 终端流操作符，在[作用域]中[启动][启动]给定流的[收集][收集]。
    3.  * 它是“范围”(scope)的简称。启动{flow.collect()} '。
    4.  *
    5.  * 此操作符通常与[onEach]， [onCompletion]和[catch]操作符一起使用，以处理所有发出的值
    6.  * 处理上游流或处理过程中可能发生的异常，例如:
    7.  *
    8.  * ```
    9.  * flow
    10.  *     .onEach { value -> updateUi(value) }
    11.  *     .onCompletion { cause -> updateUi(if (cause == null) "Done" else "Failed") }
    12.  *     .catch { cause -> LOG.error("Exception: $cause") }
    13.  *     .launchIn(uiScope)
    14.  * ```
    15.  *
    16.  * 注意，[launchIn]的结果值没有被使用，提供的作用域负责取消。
    17.  */
    18. public fun  Flow.launchIn(scope: CoroutineScope): Job = scope.launch {
    19.     collect() // tail-call
    20. }

    3、代码示例

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.   runBlocking {
    6.            val mFlowEvent: Flow<Int> = flowEvent()
    7.            val mOnEachFlow:Flow<Int> = mFlowEvent.onEach (action={
    8.                // 逐个处理产生的事件
    9.                Log.e(TAG, "收集Flow异步流冷流mFlowEvent中的一个个元素it=$it -------- mFlow的协程所在的上下文 : ${Thread.currentThread().name}")
    10.            })
    11.            // 该 launchIn 函数返回一个 Job 对象
    12.            val job: Job =   mOnEachFlow.launchIn(CoroutineScope(Dispatchers.IO)) // 在指定的协程作用域中处理收集元素操作,
    13.                job.join()  // 该协程不是 runBlocking 主协程 的子协程, 需要调用 join 等待协程执行完毕
    14.        }
    15.  
    16.   /**
    17.    * Flow 异步流冷流mAsFlow发射元素值it=0 -------- mAsFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    18.    * 收集Flow异步流冷流mFlowEvent中的一个个元素it=0 -------- mFlow的协程所在的上下文 : DefaultDispatcher-worker-3
    19.    *
    20.    * Flow 异步流冷流mAsFlow发射元素值it=1 -------- mAsFlow构建器所在的上下文 : DefaultDispatcher-worker-3
    21.    * 收集Flow异步流冷流mFlowEvent中的一个个元素it=1 -------- mFlow的协程所在的上下文 : DefaultDispatcher-worker-1
    22.    *
    23.    * Flow 异步流冷流mAsFlow发射元素值it=2 -------- mAsFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    24.    * 收集Flow异步流冷流mFlowEvent中的一个个元素it=2 -------- mFlow的协程所在的上下文 : DefaultDispatcher-worker-2
    25.    *
    26.    * Flow 异步流冷流mAsFlow发射元素值it=3 -------- mAsFlow构建器所在的上下文 : DefaultDispatcher-worker-2
    27.    *收集Flow异步流冷流mFlowEvent中的一个个元素it=3 -------- mFlow的协程所在的上下文 : DefaultDispatcher-worker-1
    28.    * */
    29.  
    30.  
    31. }
    32.  
    33.  
    34.   /**
    35.      * 使用 flow 构建器 Flow 异步流
    36.      * 产生事件的 事件源
    37.      */
    38.  
    39.     suspend fun  flowEvent(): Flow<Int>{
    40.         // 将区间转为 Flow 流
    41.         val mAsFlow: Flow<Int>  = (0..3).asFlow()
    42.         val mOnEachFlow:Flow<Int> = mAsFlow.onEach(action={
    43.             // 发射元素 ( 产生事件 ) 时挂起 500ms
    44.             delay(500)
    45.             Log.e(TAG,"Flow 异步流冷流mAsFlow发射元素值it=$it -------- mAsFlow构建器所在的上下文 : ${Thread.currentThread().name}")
    46.         })
    47.  
    48.        val mFlowOn:Flow<Int> =  mOnEachFlow.flowOn(context=Dispatchers.Default)           // 设置发射元素的协程
    49.         return  mFlowOn
    50.     }
    51.  
    52. }

    通过取消流收集所在的协程取消流

    Flow 流的 收集元素 操作 , 是在协程中执行 , 将 协程 取消 , 即可将 Flow 流收集操作 取消 , 也就是 将 Flow 流取消 ;

    在 Flow#collect 代码块中 , 执行 Job#cancel 函数 , 即可取消该流收集操作所在的协程 , 进而取消了流 ;

     代码示例1 :使用 withTimeoutOrNull(2000) 创建一个协程 , 该协程在 3000ms 后自动超时取消 , 同时在其中进行 流收集 的操作也一并取消 ;

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.  runBlocking {
    6.  
    7.       // 该协程作用域 2 秒后超时取消
    8.       withTimeoutOrNull(timeMillis=3000,block={
    9.           // 协程中调用挂起函数返回一个 Flow 异步流
    10.           val mFlowEvent: Flow<Int> = flowEvent()
    11.           mFlowEvent.collect(collector={
    12.               Log.e(TAG, "收集Flow异步流冷流mFlowEvent中的一个个元素it=$it -------- mFlowEvent的协程所在的上下文 : ${Thread.currentThread().name}")
    13.           })
    14.       })
    15.       Log.e(TAG,"协程作用域取消")
    16.  
    17.   }
    18.  
    19.         /**Flow 异步流冷流mAsFlow发射元素值it=0 -------- mAsFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    20.          * 收集Flow异步流冷流mFlowEvent中的一个个元素it=0 -------- mFlowEvent的协程所在的上下文 : main
    21.          *
    22.          * Flow 异步流冷流mAsFlow发射元素值it=1 -------- mAsFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    23.          * 收集Flow异步流冷流mFlowEvent中的一个个元素it=1 -------- mFlowEvent的协程所在的上下文 : main
    24.          *
    25.          * Flow 异步流冷流mAsFlow发射元素值it=2 -------- mAsFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    26.          * 收集Flow异步流冷流mFlowEvent中的一个个元素it=2 -------- mFlowEvent的协程所在的上下文 : main
    27.          *
    28.          * Flow 异步流冷流mAsFlow发射元素值it=3 -------- mAsFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    29.          * 收集Flow异步流冷流mFlowEvent中的一个个元素it=3 -------- mFlowEvent的协程所在的上下文 : main
    30.          *
    31.          * Flow 异步流冷流mAsFlow发射元素值it=4 -------- mAsFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    32.          * 收集Flow异步流冷流mFlowEvent中的一个个元素it=4 -------- mFlowEvent的协程所在的上下文 : main
    33.          *
    34.          * 协程作用域取消
    35.          * */
    36.  
    37. }
    38.  
    39.   /**
    40.      * 使用 flow 构建器 Flow 异步流
    41.      * 产生事件的 事件源
    42.      */
    43.  
    44.     private suspend fun  flowEvent(): Flow<Int>{
    45.         // 将区间转为 Flow 流
    46.         val mAsFlow: Flow<Int>  = (0..10).asFlow()
    47.         val mOnEachFlow:Flow<Int> = mAsFlow.onEach(action={
    48.             // 发射元素 ( 产生事件 ) 时挂起 500ms
    49.             delay(500)
    50.             Log.e(TAG,"Flow 异步流冷流mAsFlow发射元素值it=$it -------- mAsFlow构建器所在的上下文 : ${Thread.currentThread().name}")
    51.         })
    52.  
    53.        val mFlowOn:Flow<Int> =  mOnEachFlow.flowOn(context=Dispatchers.Default)           // 设置发射元素的协程
    54.         return  mFlowOn
    55.     }
    56.  
    57.  
    58. }

    调用 FlowCollector#emit 发射元素时自动执行 Flow 流的取消检测

    在 Flow 流构建器 中 , 每次 调用 FlowCollector#emit 发射元素时 ,都会执行一个 ensureActive 检测 , 检测当前的流是否取消 ,因此 , 在 flow 流构建器 中 , 循环执行的 FlowCollector#emit 发射操作 , 是可以取消的 ;

    在 Flow#collect 代码块中 , 执行 Job#cancel 函数 , 即可取消该流收集操作所在的协程 , 进而取消了流 ;

    1. /**
    2.  * 用一个可选的cancel [cause]取消这个作用域，包括它的作业和它的所有子任务。
    3.  * 原因可用于指定错误消息或提供有关的其他详细信息
    4.  * 用于调试目的的取消原因。
    5.  * 如果作用域中没有作业，则抛出[IllegalStateException]。
    6.  */
    7. public fun CoroutineScope.cancel(cause: CancellationException? = null) {
    8.     val job = coroutineContext[Job] ?: error("Scope cannot be cancelled because it does not have a job: $this")
    9.     job.cancel(cause)
    10. }

     

     代码示例1 :在 Flow#collect 代码块中 , 执行 Job#cancel 函数 , 即可取消该流收集操作所在的协程 , 进而取消了流 ;

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.  runBlocking {
    6.                  val mFlowEvent2 : Flow<Int> = flowEvent2()
    7.                mFlowEvent2.collect(collector={
    8.                    Log.e(TAG, "收集Flow异步流冷流mFlowEvent2中的一个个元素it=$it -------- mFlowEvent2的协程所在的上下文 : ${Thread.currentThread().name}")
    9.                    if(it==3){
    10.                        // 收集到元素 3 时, 取消流
    11.                        // 在流中 emit 发射 3 时, 就会自动爆出异常, 停止后续操作
    12.                        cancel()
    13.                    }
    14.                })
    15.            }
    16.  
    17.         /**
    18.          *收集Flow异步流冷流mFlowEvent2中的一个个元素it=1 -------- mFlowEvent2的协程所在的上下文 : main
    19.          *Flow 异步流冷流mFlow发射元素值i=1 --------mFlow构建器所在的上下文 : main
    20.          *
    21.          * 收集Flow异步流冷流mFlowEvent2中的一个个元素it=2 -------- mFlowEvent2的协程所在的上下文 : main
    22.          * Flow 异步流冷流mFlow发射元素值i=2 --------mFlow构建器所在的上下文 : main
    23.          *
    24.          * 收集Flow异步流冷流mFlowEvent2中的一个个元素it=3 -------- mFlowEvent2的协程所在的上下文 : main
    25.          * Flow 异步流冷流mFlow发射元素值i=3 --------mFlow构建器所在的上下文 : main
    26.          */
    27.  
    28. }
    29.  
    30.   /**
    31.      * 使用 flow 构建器 Flow 异步流
    32.      */
    33.     private suspend fun  flowEvent2(): Flow<Int>{
    34.         val mFlow:Flow<Int> = flow<Int>(block = {
    35.             for(i in 1..6) {
    36.                 delay(1000)
    37.                 emit(i)
    38.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i --------mFlow构建器所在的上下文 : ${Thread.currentThread().name}")
    39.             }
    40.         })
    41.       return  mFlow
    42.      }
    43.  
    44.  
    45.  
    46. }

    调用 Flow#cancellable() 函数启用检测 Flow 流的取消

    在 Flow 流中 , 除 FlowCollector#emit 发射元素 之外 ,还有很多其它的 流操作 , 这些操作不会 自动执行 ensureActive 检测 ,因此这里需要我们 手动 进行 流取消检测 ;调用 Flow#cancellable() 函数 , 可以手动设置流取消检测 ;

     代码示例1 :调用 Flow#cancellable() 函数 , 可以手动设置流取消检测 ;

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.   runBlocking {
    6.  
    7.             (0..5).asFlow().cancellable().collect {
    8.                 Log.e(TAG, "收集Flow异步流冷流中的一个个元素it=$it --------流的协程所在的上下文 : ${Thread.currentThread().name}")
    9.  
    10.                 // 收集到元素 2 时, 协程退出
    11.                 if (it == 2) {
    12.                     cancel()
    13.                 }
    14.             }
    15.             Log.e(TAG,"协程作用域取消")
    16.         }
    17.  
    18.         /**
    19.          * 收集Flow异步流冷流中的一个个元素it=0 --------流的协程所在的上下文 : main
    20.          *  收集Flow异步流冷流中的一个个元素it=1 --------流的协程所在的上下文 : main
    21.          * 收集Flow异步流冷流中的一个个元素it=2 --------流的协程所在的上下文 : main
    22.          *
    23.          * */
    24. }
    25.  
    26. }

    背压概念

    " 背压 " 概念 指的是 数据 受到 与 流动方向 一致的压力 ,

    数据 生产者 的 生产效率 大于 数据 消费者 的 消费效率 , 就会产生 背压 ;

    处理背压问题 , 有 2 种方案 :

    • 降低 数据 生产者 的生产效率 ;
    • 提高 数据 消费者 的消费效率 ;

    背压代码示例 :

    以 100 ms间隔发射元素 , 以 200 ms 间隔收集元素 , 发射元素的效率 高于 收集元素的效率, 此时会产生背压 ;

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.  runBlocking {
    6.              val delta = measureTimeMillis {
    7.                  // 以 200 ms 的间隔收集元素
    8.                  // 发射元素的效率 高于 收集元素的效率, 此时会产生背压
    9.                  flowEmit().collect{
    10.                      delay(200)
    11.                      Log.e(TAG, "收集Flow异步流冷流中的一个个元素it=$it --------流的协程所在的上下文 : ${Thread.currentThread().name}")
    12.                  }
    13.              }
    14.  
    15.              Log.e(TAG,"Flow异步流冷流收集元素总共耗时 $delta ms")
    16.          }
    17.  
    18.         /**
    19.          *收集Flow异步流冷流中的一个个元素it=0 --------流的协程所在的上下文 : main
    20.          *Flow 异步流冷流mFlow发射元素值i=0 --------mFlow构建器所在的上下文 : main
    21.          收集Flow异步流冷流中的一个个元素it=1 --------流的协程所在的上下文 : main
    22.          Flow 异步流冷流mFlow发射元素值i=1 --------mFlow构建器所在的上下文 : main
    23.         收集Flow异步流冷流中的一个个元素it=2 --------流的协程所在的上下文 : main
    24.         Flow 异步流冷流mFlow发射元素值i=2 --------mFlow构建器所在的上下文 : main
    25.         收集Flow异步流冷流中的一个个元素it=3 --------流的协程所在的上下文 : main
    26.         Flow 异步流冷流mFlow发射元素值i=3 --------mFlow构建器所在的上下文 :
    27.         main收集Flow异步流冷流中的一个个元素it=4 --------流的协程所在的上下文 : main
    28.         Flow 异步流冷流mFlow发射元素值i=4--------mFlow构建器所在的上下文 : main
    29.         main收集Flow异步流冷流中的一个个元素it=5 --------流的协程所在的上下文 : main
    30.         Flow 异步流冷流mFlow发射元素值i=5--------mFlow构建器所在的上下文 : main
    31.         Flow异步流冷流收集元素总共耗时 1815 ms
    34.          * */
    35.  
    36. }
    37.  
    38.    /**
    39.      * 使用 flow 构建器 Flow 异步流
    40.      */
    41.     suspend fun flowEmit(): Flow<Int> {
    42.         // 以 100 ms 的间隔发射元素
    43.  
    44.         val mFlow:Flow<Int> = flow<Int>(block ={
    45.             for (i in 0..5) {
    46.                 delay(100)
    47.                 emit(i)
    48.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i --------mFlow构建器所在的上下文 : ${Thread.currentThread().name}")
    49.             }
    50.         })
    51.  
    52.         return  mFlow
    53.     }
    54.  
    55.  
    56.  
    57. }

    使用缓冲处理背压问题

    调用 Flow#buffer 函数 , 为 收集元素 添加一个缓冲 , 可以指定缓冲区个数 ;

    代码示例 :发射元素后 , 将发射的元素缓存起来 , 然后慢慢接收元素 ;

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.   runBlocking {
    6.             val delta = measureTimeMillis {
    7.                 // 以 200 ms 的间隔收集元素
    8.                 // 发射元素的效率 高于 收集元素的效率, 此时会产生背压
    9.                 flowEmit().buffer(10).collect{
    10.                     delay(200)
    11.                     Log.e(TAG, "收集Flow异步流冷流中的一个个元素it=$it --------流的协程所在的上下文 : ${Thread.currentThread().name}")
    12.                 }
    13.             }
    14.  
    15.             Log.e(TAG,"Flow异步流冷流收集元素总共耗时 $delta ms")
    16.         }
    17.  
    18.         /**
    19.          * Flow 异步流冷流mFlow发射元素值i=0 --------mFlow构建器所在的上下文 : main
    20.          *Flow 异步流冷流mFlow发射元素值i=1 --------mFlow构建器所在的上下文 : main
    21.          *  收集Flow异步流冷流中的一个个元素it=0 --------流的协程所在的上下文 : main
    22.          *
    23.          * Flow 异步流冷流mFlow发射元素值i=2 --------mFlow构建器所在的上下文 : main
    24.          * Flow 异步流冷流mFlow发射元素值i=3 --------mFlow构建器所在的上下文 : main
    25.          * 收集Flow异步流冷流中的一个个元素it=1 --------流的协程所在的上下文 : main
    26.          *
    27.          * Flow 异步流冷流mFlow发射元素值i=4 --------mFlow构建器所在的上下文 : main
    28.          * Flow 异步流冷流mFlow发射元素值i=5 --------mFlow构建器所在的上下文 : main
    29.          * 收集Flow异步流冷流中的一个个元素it=2 --------流的协程所在的上下文 : main
    30.          *
    31.          * 收集Flow异步流冷流中的一个个元素it=3 --------流的协程所在的上下文 : main
    32.          * 收集Flow异步流冷流中的一个个元素it=4 --------流的协程所在的上下文 : main
    33.          * 收集Flow异步流冷流中的一个个元素it=5 --------流的协程所在的上下文 : main
    34.          * Flow异步流冷流收集元素总共耗时 1320 ms
    35.          *
    36.          * */
    37.  
    38.  
    39. }
    40.  
    41.   /**
    42.      * 使用 flow 构建器 Flow 异步流
    43.      */
    44.     suspend fun flowEmit(): Flow<Int> {
    45.         // 以 100 ms 的间隔发射元素
    46.  
    47.         val mFlow:Flow<Int> = flow<Int>(block ={
    48.             for (i in 0..5) {
    49.                 delay(100)
    50.                 emit(i)
    51.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i --------mFlow构建器所在的上下文 : ${Thread.currentThread().name}")
    52.             }
    53.         })
    54.  
    55.         return  mFlow
    56.     }
    57.  
    58.  
    59.  
    60. }

    使用 flowOn 处理背压问题

    上述 发射元素 和 收集元素 都是在同一个线程中执行的 , 这两个操作可以并行执行 , 即使用 flowOn 指定收集元素的线程 ;使用 flowOn 更改了协程上下文 , 使得 发射元素 与 收集元素 在不同的线程中并行执行 ;

    代码示例 :

    1.  
    2. class FlowActivity  : ComponentActivity() {
    3.  
    4.  override   fun onCreate(savedInstanceState: Bundle?){
    5.  
    6.   runBlocking {
    7.             val delta = measureTimeMillis {
    8.                 // 以 200 ms 的间隔收集元素
    9.                 // 发射元素的效率 高于 收集元素的效率, 此时会产生背压
    10.                 flowEmit().flowOn(Dispatchers.Default).collect{
    11.                     delay(200)
    12.                     Log.e(TAG, "收集Flow异步流冷流中的一个个元素it=$it --------流的协程所在的上下文 : ${Thread.currentThread().name}")
    13.                 }
    14.             }
    15.  
    16.             Log.e(TAG,"Flow异步流冷流收集元素总共耗时 $delta ms")
    17.         }
    18.  
    19.         /**Flow 异步流冷流mFlow发射元素值i=0 --------mFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    20.          * Flow 异步流冷流mFlow发射元素值i=1 --------mFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    21.          * 收集Flow异步流冷流中的一个个元素it=0 --------流的协程所在的上下文 : main
    22.          *
    23.          * Flow 异步流冷流mFlow发射元素值i=2 --------mFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    24.          * Flow 异步流冷流mFlow发射元素值i=3 --------mFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    25.          *  收集Flow异步流冷流中的一个个元素it=1 --------流的协程所在的上下文 : main
    26.          *
    27.          *  Flow 异步流冷流mFlow发射元素值i=4 --------mFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    28.          *  Flow 异步流冷流mFlow发射元素值i=5 --------mFlow构建器所在的上下文 : DefaultDispatcher-worker-1
    29.          * 收集Flow异步流冷流中的一个个元素it=2 --------流的协程所在的上下文 : main
    30.          * 收集Flow异步流冷流中的一个个元素it=3 --------流的协程所在的上下文 : main
    31.          * 收集Flow异步流冷流中的一个个元素it=4 --------流的协程所在的上下文 : main
    32.          * 收集Flow异步流冷流中的一个个元素it=5 --------流的协程所在的上下文 : main
    33.          * Flow异步流冷流收集元素总共耗时 1316 ms
    34.          * */
    35.  
    36. }
    37.  
    38.     /**
    39.      * 使用 flow 构建器 Flow 异步流
    40.      */
    41.     suspend fun flowEmit(): Flow<Int> {
    42.         // 以 100 ms 的间隔发射元素
    43.  
    44.         val mFlow:Flow<Int> = flow<Int>(block ={
    45.             for (i in 0..5) {
    46.                 delay(100)
    47.                 emit(i)
    48.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i --------mFlow构建器所在的上下文 : ${Thread.currentThread().name}")
    49.             }
    50.         })
    51.  
    52.         return  mFlow
    53.     }
    54.  
    55.  
    56. }

    从提高收集元素效率方向解决背压问题

    从提高收集元素效率方向解决背压问题 :

    ❶调用 Flow#conflate 函数 , 合并发射元素项 , 不对每个值进行单独处理 ;

    ❷调用 Flow#collectLatest 函数 , 取消并重新发射最后一个元素 , 只关心最后一个结果 , 不关心中间的过程值 ;

    Flow#conflate 代码示例:

    发射了 6 个元素 , 但是只接收到了 4个元素 , 接收的元素 2  4 被过滤掉了 ;

    1.  
    2. class FlowActivity  : ComponentActivity() {
    3.  
    4.  override   fun onCreate(savedInstanceState: Bundle?){
    5.  
    6.   runBlocking {
    7.             val delta = measureTimeMillis {
    8.                 // 以 200 ms 的间隔收集元素
    9.                 // 发射元素的效率 高于 收集元素的效率, 此时会产生背压
    10.                 flowEmit().conflate().collect{
    11.                     delay(200)
    12.                     Log.e(TAG, "收集Flow异步流冷流中的一个个元素it=$it --------流的协程所在的上下文 : ${Thread.currentThread().name}")
    13.                 }
    14.             }
    15.  
    16.             Log.e(TAG,"Flow异步流冷流收集元素总共耗时 $delta ms")
    17.         }
    18.  
    19. /**
    20.  *Flow 异步流冷流mFlow发射元素值i=0 --------mFlow构建器所在的上下文 : main
    21.  *  Flow 异步流冷流mFlow发射元素值i=1 --------mFlow构建器所在的上下文 : main
    22.  * 收集Flow异步流冷流中的一个个元素it=0 --------流的协程所在的上下文 : main
    23.  *
    24.  *  Flow 异步流冷流mFlow发射元素值i=2 --------mFlow构建器所在的上下文 : main
    25.  *   Flow 异步流冷流mFlow发射元素值i=3 --------mFlow构建器所在的上下文 : main
    26.  * 收集Flow异步流冷流中的一个个元素it=1 --------流的协程所在的上下文 : main
    27.  *
    28.  * Flow 异步流冷流mFlow发射元素值i=4 --------mFlow构建器所在的上下文 : main
    29.  * Flow 异步流冷流mFlow发射元素值i=5 --------mFlow构建器所在的上下文 : main
    30.  * 收集Flow异步流冷流中的一个个元素it=3 --------流的协程所在的上下文 : main
    31.  * 收集Flow异步流冷流中的一个个元素it=5 --------流的协程所在的上下文 : main
    32.  *
    33.  * Flow异步流冷流收集元素总共耗时 915 ms
    34.  *
    35.  * */
    36.  
    37.  
    38. }
    39.  
    40.  /**
    41.      * 使用 flow 构建器 Flow 异步流
    42.      */
    43.     suspend fun flowEmit(): Flow<Int> {
    44.         // 以 100 ms 的间隔发射元素
    45.  
    46.         val mFlow:Flow<Int> = flow<Int>(block ={
    47.             for (i in 0..5) {
    48.                 delay(100)
    49.                 emit(i)
    50.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i --------mFlow构建器所在的上下文 : ${Thread.currentThread().name}")
    51.             }
    52.         })
    53.  
    54.         return  mFlow
    55.     }
    56.  
    57.  
    58. }

    Flow#collectLatest 代码示例:

     只接收了最后一个元素 , 前几个元素没有接收 ;

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.   runBlocking {
    6.             val delta = measureTimeMillis {
    7.                 // 以 200 ms 的间隔收集元素
    8.                 // 发射元素的效率 高于 收集元素的效率, 此时会产生背压
    9.                 flowEmit().collectLatest {
    10.                     delay(200)
    11.                     Log.e(TAG, "收集Flow异步流冷流中的一个个元素it=$it --------流的协程所在的上下文 : ${Thread.currentThread().name}")
    12.                 }
    13.             }
    14.  
    15.             Log.e(TAG,"Flow异步流冷流收集元素总共耗时 $delta ms")
    16.         }
    17.  
    18.         /**Flow 异步流冷流mFlow发射元素值i=0 --------mFlow构建器所在的上下文 : main
    19.          * Flow 异步流冷流mFlow发射元素值i=1 --------mFlow构建器所在的上下文 : main
    20.          * Flow 异步流冷流mFlow发射元素值i=2 --------mFlow构建器所在的上下文 : main
    21.          * Flow 异步流冷流mFlow发射元素值i=3 --------mFlow构建器所在的上下文 : main
    22.          * Flow 异步流冷流mFlow发射元素值i=4 --------mFlow构建器所在的上下文 : main
    23.          * Flow 异步流冷流mFlow发射元素值i=5 --------mFlow构建器所在的上下文 : main
    24.          *
    25.          * 收集Flow异步流冷流中的一个个元素it=5 --------流的协程所在的上下文 : main
    26.          * Flow异步流冷流收集元素总共耗时 838 ms
    27.            */
    28.  
    29.  
    30. }
    31.  
    32.   /**
    33.      * 使用 flow 构建器 Flow 异步流
    34.      */
    35.     suspend fun flowEmit(): Flow<Int> {
    36.         // 以 100 ms 的间隔发射元素
    37.  
    38.         val mFlow:Flow<Int> = flow<Int>(block ={
    39.             for (i in 0..5) {
    40.                 delay(100)
    41.                 emit(i)
    42.                 Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i --------mFlow构建器所在的上下文 : ${Thread.currentThread().name}")
    43.             }
    44.         })
    45.  
    46.         return  mFlow
    47.     }
    48.  
    49.  
    50. }

    Flow 操作符

    Flow 操作符主要分类:过渡操作符、限长操作符、末端操作符

    过渡操作符

    过渡操作符 相关概念 :

    ❶转换流 : 使用 过渡操作符 转换 Flow 流 ;
    ❷作用位置 : 过渡操作符作用 于 流的上游 , 返回 流的下游 ;
    ❸非挂起函数 : 过渡操作符 不是挂起函数 , 属于冷操作符 ;    
    ❹运行速度 : 过渡操作符 可以 快速返回 新的 转换流 ;

    1.map 操作符

    通过 map 操作符 , 可以操作每个元素 , 将元素转为另外一种类型的元素 ;

    map 操作符原型 :
    1. /**
    2.  * 返回一个流，其中包含对原始流的每个值应用给定[transform]函数的结果。
    3.  */
    4. public inline fun  Flow.map(crossinline transform: suspend (value: T) -> R): Flow = transform { value ->
    5.    return@transform emit(transform(value))
    6. }
    代码示例 : 将 Flow 中发射的 Int 元素 转为 字符串 ; 通过 map 操作符 , 将 Int 类型的元素 转为 字符串类型 元素 ;
    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.   runBlocking {
    6.            (0..3).asFlow()
    7.                // 通过 map 操作符  将 Flow 中发射的 Int 元素it 转为 字符串
    8.                .map(transform={
    9.                  stringConvert(it)
    10.            })
    11.                .collect(collector={
    12.                    Log.e(TAG, "收集Flow异步流冷流中的一个个元素it=$it --------流的协程所在的上下文 : ${Thread.currentThread().name}")
    13.                })
    14.        }
    15.  
    16.         /**
    17.          * 收集Flow异步流冷流中的一个个元素it=convert 0 --------流的协程所在的上下文 : main
    18.          * 收集Flow异步流冷流中的一个个元素it=convert 1 --------流的协程所在的上下文 : main
    19.          * 收集Flow异步流冷流中的一个个元素it=convert 2 --------流的协程所在的上下文 : main
    20.          * 收集Flow异步流冷流中的一个个元素it=convert 3 --------流的协程所在的上下文 : main
    21.          * */
    22.  
    23.  
    24.  
    25. }
    26.  
    27.   // 将 Int 转为 字符串
    28.     suspend fun stringConvert(num: Int): String {
    29.         delay(1000)
    30.         return "convert $num"
    31.     }
    32.  
    33.  
    34. }

    2.transform 操作符

    通过 transform 操作符 , 可以操作每个元素 , 可以在单个元素处理时 , 发射多次元素 ;

    transform 操作符原型 :
    1. /**
    2.  * 将[transform]函数应用到给定流的每个值。
    3.  *
    4.  * ' transform '的接收者是[FlowCollector]，因此' transform '是一个
    5.  * 灵活的函数，可以转换发出的元素，跳过它或多次发出它。
    6.  *
    7.  * 该操作符泛化了[filter]和[map]操作符和
    8.  * 可以用作其他操作符的构建块，例如:
    9.  *
    10.  * ```
    11.  * fun Flow.skipOddAndDuplicateEven(): Flow = transform { value ->
    12.  *     if (value % 2 == 0) { // Emit only even values, but twice
    13.  *         emit(value)
    14.  *         emit(value)
    15.  *     } // Do nothing if odd
    16.  * }
    17.  * ```
    18.  */
    19. public inline fun  Flow.transform(
    20.     @BuilderInference crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
    21. ): Flow = flow { // 注意:这里使用的是安全流，因为收集器对每个操作的转换都是公开的
    22.     collect { value ->
    23.         // 没有它，单元将被退回，TCE将不会生效，KT-28938
    24.         return@collect transform(value)
    25.     }
    26. }
    代码示例 :
    1.  
    2. class FlowActivity  : ComponentActivity() {
    3.  
    4.  override   fun onCreate(savedInstanceState: Bundle?){
    5.  
    6.    runBlocking {
    7.             (0..3).asFlow()
    8.                 // 通过 map 操作符  将 Flow 中发射的 Int 元素it 转为 字符串
    9.                 .transform (transform={
    10.                     Log.e(TAG,"接受者对象this=$this,回调的参数it=$it")
    11.                         // 在 transform 操作符中发射 2 个元素
    12.                     emit(it)
    13.                     emit(stringConvert(it))
    14.  
    15.                 })
    16.                 .collect(collector={
    17.                     Log.e(TAG, "收集Flow异步流冷流中的一个个元素it=$it --------流的协程所在的上下文 : ${Thread.currentThread().name}")
    18.                 })
    19.         }
    20.  
    21.  
    22.         /**
    23.          *收集Flow异步流冷流中的一个个元素it=0 --------流的协程所在的上下文 : main
    24.          * 收集Flow异步流冷流中的一个个元素it=convert 0 --------流的协程所在的上下文 : main
    25.          * 接受者对象this=Continuation at kotlinx.coroutines.flow.internal.SafeCollector,回调的参数it=1
    26.          *
    27.          * 收集Flow异步流冷流中的一个个元素it=1 --------流的协程所在的上下文 : main
    28.          * 收集Flow异步流冷流中的一个个元素it=convert 1 --------流的协程所在的上下文 : main
    29.          * 接受者对象this=Continuation at kotlinx.coroutines.flow.internal.SafeCollector,回调的参数it=2
    30.          *
    31.          * 收集Flow异步流冷流中的一个个元素it=2 --------流的协程所在的上下文 : main
    32.          *收集Flow异步流冷流中的一个个元素it=convert 2 --------流的协程所在的上下文 : main
    33.          * 接受者对象this=Continuation at kotlinx.coroutines.flow.internal.SafeCollector,回调的参数it=3
    34.          *
    35.          * 收集Flow异步流冷流中的一个个元素it=3 --------流的协程所在的上下文 : main
    36.          * 收集Flow异步流冷流中的一个个元素it=convert 3 --------流的协程所在的上下文 : main
    37.          *
    38.          * */
    39.  
    40.  
    41.  
    42. }
    43.  
    44.   // 将 Int 转为 字符串
    45.     suspend fun stringConvert(num: Int): String {
    46.         delay(1000)
    47.         return "convert $num"
    48.     }
    49.  
    50.  
    51.  
    52. }

    限长操作符

    1.take 操作符

    通过 take 操作符 , 可以选择选取指定个数的发射元素 ;

    如 : 在 Flow 流中发射了 4 个元素 , 但是调用了 Flow#take(2) , 只收集其中 2 个元素 ;

     take 操作符原型 :
    1. /**
    2.  * 返回包含第一个[count]元素的流。
    3.  * 当[count]元素被消耗时，原始流将被取消。
    4.  * 如果[count]不是正数，抛出[IllegalArgumentException]。
    5.  */
    6. public fun  Flow.take(count: Int): Flow {
    7.     require(count > 0) { "Requested element count $count should be positive" }
    8.     return unsafeFlow {
    9.         var consumed = 0
    10.         try {
    11.             collect { value ->
    12.                 // 注意:这个for take不是故意用collectWhile写的。
    13.                 // 它首先检查条件，然后对emit或emitAbort进行尾部调用。
    14.                 // 这样，正常的执行不需要状态机，只需要终止(emitAbort)。
    15.                 // 有关不同方法的比较，请参阅“TakeBenchmark”。
    16.                 if (++consumed < count) {
    17.                     return@collect emit(value)
    18.                 } else {
    19.                     return@collect emitAbort(value)
    20.                 }
    21.             }
    22.         } catch (e: AbortFlowException) {
    23.             e.checkOwnership(owner = this)
    24.         }
    25.     }
    26. }
    代码示例 :

     

    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5.  runBlocking {
    6.             (0..5).asFlow()
    7.                 .take(3)
    8.                 .collect(collector={
    9.                     Log.e(TAG, "收集Flow异步流冷流中的一个个元素it=$it --------流的协程所在的上下文 : ${Thread.currentThread().name}")
    10.                 })
    11.         }
    12.  
    13.         /**
    14.          * 收集Flow异步流冷流中的一个个元素it=0 --------流的协程所在的上下文 : main
    15.          * 收集Flow异步流冷流中的一个个元素it=1 --------流的协程所在的上下文 : main
    16.          * 收集Flow异步流冷流中的一个个元素it=2 --------流的协程所在的上下文 : main
    17.          * 
    18.          * */
    19.  
    20. }
    21.  
    22. }

    末端(终端)操作符

    何为末端(终端)操作符函数? 末端(终端)操作符的后面不可以再接其他操作符函数了，而是只能获取最终的运行结果。

    末端操作符 指的是 在 Flow 流最末端 调用 挂起函数 收集元素 的操作符 , 最常见的 末端操作符 就是 collect 操作符 ;

    常见的末端(终端)操作符 :

    • 收集元素 : collect ;
    • 将收集的元素转为集合 : toList , toSet ;
    • 收集第一个元素 : first ;
    • 发射单个元素 : single ;
    • 规约流到单个值 : reduce , fold ;

    1、collect 操作符

      collect 操作符原型 :
    1. /**
    2.  * 终端流操作符，使用提供的[动作]收集给定的流。
    3.  * 如果在收集过程中或在所提供的流中发生任何异常，则此方法将重新抛出此异常。
    4.  *
    5.  * 使用示例:
    6.  *
    7.  * ```
    8.  * val flow = getMyEvents()
    9.  * try {
    10.  *     flow.collect { value ->
    11.  *         println("Received $value")
    12.  *     }
    13.  *     println("My events are consumed successfully")
    14.  * } catch (e: Throwable) {
    15.  *     println("Exception from the flow: $e")
    16.  * }
    17.  * ```
    18.  */
    19. public suspend inline fun  Flow.collect(crossinline action: suspend (value: T) -> Unit): Unit =
    20.     collect(object : FlowCollector {
    21.         override suspend fun emit(value: T) = action(value)
    22.     })

    2、reduce 操作符

    我个人认为reduce函数还是比较好理解的，它的基本公式如下：

    flow.reduce { acc, value -> acc + value }

    其中acc是累积值的意思，value则是当前值的意思。

    也就是说，reduce函数会通过参数给我们一个Flow的累积值acc和一个Flow的当前值value ，我们可以在函数体中对它们进行一定的运算，运算的结果会作为下一个累积值继续传递到reduce函数当中。

    reduce 操作符原型 :

    从第一个元素开始累加值，并对当前累加器值和每个元素应用[操作]。

    1. /**
    2.  * 从第一个元素开始累加值，并对当前累加器值和每个元素应用[操作]。
    3.  * 如果流为空，抛出[NoSuchElementException]。
    4.  */
    5. public suspend fun  Flow.reduce(operation: suspend (accumulator: S, value: T) -> S): S {
    6.     var accumulator: Any? = NULL
    7.  
    8.     collect { value ->
    9.         accumulator = if (accumulator !== NULL) {
    10.             @Suppress("UNCHECKED_CAST")
    11.             operation(accumulator as S, value)
    12.         } else {
    13.             value
    14.         }
    15.     }
    16.  
    17.     if (accumulator === NULL) throw NoSuchElementException("Empty flow can't be reduced")
    18.     @Suppress("UNCHECKED_CAST")
    19.     return accumulator as S
    20. }

    reduce使用示例:

    举一个更加具体点的例子，我们上学时学等差数列都会讲这个故事，高斯的老师让全班同学计算从1加到100的结果。今天我们不需要借助等差数列，只需要借助reduce函数就可以立刻算出结果了：

    reduce函数是一个终端操作符函数，它的后面不可以再接其他操作符函数了，而是只能获取最终的运行结果。

    1.  
    2. class FlowActivity  : ComponentActivity() {
    3.  
    4.  override   fun onCreate(savedInstanceState: Bundle?){
    5.  
    6.   runBlocking {
    7.  
    8.              val mFlow : Flow<Int> = flow(block={
    9.                   for (i in (1..100)) {
    10.                       emit(i)
    11.                       Log.e(TAG,"Flow 异步流冷流mFlow发射元素值i=$i --------mFlow构建器所在的上下文 : ${Thread.currentThread().name}")
    12.                   }
    13.               })
    14.           val result:Int =  mFlow.reduce(operation={ acc: Int, value:Int->
    15.                  acc+value
    16.             })
    17.  
    18.             Log.e(TAG,"返回最终计算结果result=$result")
    19.  
    20.         }
    21.  
    22.  
    23.      /**
    24.       * Flow 异步流冷流mFlow发射元素值i=1 --------mFlow构建器所在的上下文 : main
    25.       *  Flow 异步流冷流mFlow发射元素值i=2 --------mFlow构建器所在的上下文 : main
    26.       * ...
    27.       * Flow 异步流冷流mFlow发射元素值i=99 --------mFlow构建器所在的上下文 : main
    28.       * Flow 异步流冷流mFlow发射元素值i=100 --------mFlow构建器所在的上下文 : main
    29.       * 返回最终计算结果result=5050
    30.       * */
    31.  
    32. }
    33.  
    34. }
    1.    runBlocking {
    2.            val mAsFlow:Flow<Int> = (1..100)
    3.                                     .asFlow() //asFlow=flow+for
    4.             val result:Int =  mAsFlow.reduce(operation={ acc: Int, value:Int->
    5.                 acc+value
    6.             })
    7.             Log.e(TAG,"返回最终计算结果result=$result")
    8.         }
    9.  
    10.      // 返回最终计算结果result=5050

    3、fold 操作符

    从[initial]值开始累加值，并应用[operation]当前累加器值和每个元素

    fold函数和reduce函数基本上是完全类似的，

    主要的区别在于，fold函数需要传入一个初始值initial，这个初始值initial会作为首个累积值被传递到fold的函数体当中，它的基本公式如下：

    flow.fold(initial) { acc, value -> acc + value }

    注意:其实reduce函数和fold函数并不是只能用作数值计算，相反它们可以作用于任何类型的数据。

    fold 操作符原型 :
    1. /**
    2.  * 从[initial]值开始累加值，并应用[operation]当前累加器值和每个元素
    3.  */
    4. public suspend inline fun  Flow.fold(
    5.     initial: R,
    6.     crossinline operation: suspend (acc: R, value: T) -> R
    7. ): R {
    8.     var accumulator = initial
    9.     collect { value ->
    10.         accumulator = operation(accumulator, value)
    11.     }
    12.     return accumulator
    13. }
    fold使用示例:
    1. class FlowActivity  : ComponentActivity() {
    2.  
    3.  override   fun onCreate(savedInstanceState: Bundle?){
    4.  
    5. }
    6.  
    7. }

    4、single 操作符

    single 操作符原型 :

    终端操作符，等待一个且仅等待一个值发出。

    1. /**
    2.  * 终端操作符，等待一个且仅等待一个值发出。
    3.  * 为空流抛出[NoSuchElementException]，为流抛出[IllegalStateException]
    4.  * 包含多个元素的。
    5.  */
    6. public suspend fun  Flow.single(): T {
    7.     var result: Any? = NULL
    8.     collect { value ->
    9.         require(result === NULL) { "Flow has more than one element" }
    10.         result = value
    11.     }
    12.  
    13.     if (result === NULL) throw NoSuchElementException("Flow is empty")
    14.     return result as T
    15. }

    5、first 操作符

    终端操作符，返回流发出的第一个元素，然后取消流的收集。

    first 操作符原型 :
    1. /**
    2.  * 终端操作符，返回流发出的第一个元素，然后取消流的收集。
    3.  * 如果流为空，则抛出[NoSuchElementException]。
    4.  */
    5. public suspend fun  Flow.first(): T {
    6.     var result: Any? = NULL
    7.     collectWhile {
    8.         result = it
    9.         false
    10.     }
    11.     if (result === NULL) throw NoSuchElementException("Expected at least one element")
    12.     return result as T
    13. }

    鸣谢：

    mAndroid面试题之Kotlin异步流、冷流Flow (qq.com)

  • 相关阅读:
    npm常用命令详解与实践
    第九章Redis持久化
    改良版芯片重回中国市场？GPT-5 明年降临？丨 RTE 开发者日报 Vol.82
    商品720vr全景环物制作便捷推送到全世界
    视频清晰度优化指南
    rust运算
    QT连接Sqlite
    数据结构与算法之希尔排序
    前端岗位初入职场后的最初一段时间需要做什么
    docker常用命令
  • 原文地址：https://blog.csdn.net/qq_33552379/article/details/139869905