一、冷流 ( 流被收集时运行 )

Flow 异步流 的 构建器函数 flow 函数 中的 代码 ,

在 调用 Flow#collect 函数 时 , 也就是在 Flow 异步流 收集元素时 ,

才会 执行 flow 构建器 中的代码 ;

这种机制的异步流 称为 冷流 ;

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

package kim.hsl.coroutine

import android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import kotlinx.coroutines.InternalCoroutinesApi
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.AbstractFlow
import kotlinx.coroutines.flow.FlowCollector
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

class MainActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        // 携程中调用挂起函数返回一个 Flow 异步流
        runBlocking {
            println("Flow 异步流 开始收集元素")
            // 调用 Flow#collect 函数, 可以获取在异步流中产生的元素
            flowFunction().collect {
                // 每隔 500ms 即可拿到一个 Int 元素
                // 并且该操作是异步操作, 不会阻塞调用线程
                println("收集到元素 $it")
                println(it)
            }
        }
    }

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

执行结果 : 根据打印的日志可知 , 异步流收集元素后 , 才开始执行 flow 构建器 中的代码 ;

2022-12-22 16:57:31.969 26807-26807/kim.hsl.coroutine I/System.out: Flow 异步流 开始收集元素
2022-12-22 16:57:31.978 26807-26807/kim.hsl.coroutine I/System.out: Flow 异步流开始执行
2022-12-22 16:57:32.512 26807-26807/kim.hsl.coroutine I/System.out: 发射元素 0
2022-12-22 16:57:32.523 26807-26807/kim.hsl.coroutine I/System.out: 收集到元素 0
2022-12-22 16:57:32.524 26807-26807/kim.hsl.coroutine I/System.out: 0
2022-12-22 16:57:33.066 26807-26807/kim.hsl.coroutine I/System.out: 发射元素 1
2022-12-22 16:57:33.066 26807-26807/kim.hsl.coroutine I/System.out: 收集到元素 1
2022-12-22 16:57:33.066 26807-26807/kim.hsl.coroutine I/System.out: 1
2022-12-22 16:57:33.574 26807-26807/kim.hsl.coroutine I/System.out: 发射元素 2
2022-12-22 16:57:33.574 26807-26807/kim.hsl.coroutine I/System.out: 收集到元素 2
2022-12-22 16:57:33.574 26807-26807/kim.hsl.coroutine I/System.out: 2

二、流的连续性

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

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

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

代码示例 :

package kim.hsl.coroutine

import android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.runBlocking

class MainActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        // 携程中调用挂起函数返回一个 Flow 异步流
        runBlocking {
            // 使用下面的方式可以快速构建一个 Flow 流
            (0..5).asFlow().filter {
                // 奇数才能继续向下流
                // 偶数被过滤掉了
                it % 2 == 1
            }.map {
                // 遍历元素, 将其拼接成字符串
                "学号 : $it"
            }.collect {
                // 打印收集的元素
                println(it)
            }
        }
    }
}

执行结果 :

2022-12-22 17:51:46.301 5299-5299/kim.hsl.coroutine I/System.out: 学号 : 1
2022-12-22 17:51:46.301 5299-5299/kim.hsl.coroutine I/System.out: 学号 : 3
2022-12-22 17:51:46.301 5299-5299/kim.hsl.coroutine I/System.out: 学号 : 5