Flow基础 ⭐⭐

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
// 定义一个 Flow —— 此时不会执行任何代码，仅仅是"声明"
val myFlow: Flow<Int> = flow {
    println("🚀 Flow 开始执行")   // 只有 collect 时才会打印
    emit(1)                        // 发射第一个值
    delay(100)                     // 模拟异步延迟（挂起，不阻塞线程）
    emit(2)                        // 发射第二个值
    delay(100)
    emit(3)                        // 发射第三个值
    println("✅ Flow 执行完毕")
}
 
fun main() = runBlocking {
    println("--- 第一次收集 ---")
    myFlow.collect { value ->       // 第一次收集：从头执行 flow { } 块
        println("收到: $value")
    }
 
    println("--- 第二次收集 ---")
    myFlow.collect { value ->       // 第二次收集：再次从头执行，完全独立
        println("收到: $value")
    }
}

--- 第一次收集 ---
🚀 Flow 开始执行
收到: 1
收到: 2
收到: 3
✅ Flow 执行完毕
--- 第二次收集 ---
🚀 Flow 开始执行
收到: 1
收到: 2
收到: 3
✅ Flow 执行完毕

// ❶ 普通函数 —— 同步返回多个值（一次性）
fun getNumbers(): List<Int> {
    // 必须等所有值都准备好，才能一次性返回
    return listOf(1, 2, 3)
}
 
// ❷ Sequence —— 同步、惰性地逐个产生值
fun getSequence(): Sequence<Int> = sequence {
    yield(1)          // 逐个产生
    Thread.sleep(100) // ⚠️ 只能用阻塞方式等待，不能调用 suspend 函数！
    yield(2)
    yield(3)
}
 
// ❸ suspend 函数 —— 异步返回单个值
suspend fun getSingleValue(): Int {
    delay(100)        // 可以挂起，但只能返回一个值
    return 42
}
 
// ❹ Flow —— 异步 + 多值 ✅ 完美方案
fun getFlow(): Flow<Int> = flow {
    emit(1)           // 逐个发射
    delay(100)        // ✅ 可以挂起！不阻塞线程
    emit(2)
    emit(3)
}

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
// ---------- Producer（生产者）----------
fun fetchUserIds(): Flow<Int> = flow {
    val ids = listOf(101, 102, 103, 104, 105) // 模拟从数据库分批读取用户 ID
    for (id in ids) {
        delay(50)      // 模拟异步 I/O 延迟（挂起，不阻塞线程）
        emit(id)       // 将每个 ID 逐一发射到下游
        println("  [Producer] 已发射: $id")
    }
}
 
fun main() = runBlocking {
 
    fetchUserIds()                          // ① 创建 Flow（此刻什么也没发生）
        // ---------- Intermediary（中间操作）----------
        .filter { id ->                     // ② 过滤：只保留奇数 ID
            id % 2 != 0
        }
        .map { id ->                        // ③ 变换：将 ID 映射为用户名字符串
            "User-$id"
        }
        // ---------- Consumer（消费者）----------
        .collect { userName ->              // ④ 终端操作：触发整条链路执行
            println("  [Consumer] 收到: $userName")
        }
}

  [Producer] 已发射: 101
  [Consumer] 收到: User-101
  [Producer] 已发射: 102
  [Producer] 已发射: 103
  [Consumer] 收到: User-103
  [Producer] 已发射: 104
  [Producer] 已发射: 105
  [Consumer] 收到: User-105

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    flow {
        for (i in 1..3) {
            println("[Producer] 准备发射 $i (${System.currentTimeMillis()})")
            emit(i)          // 如果消费者还在忙，这里会自动挂起等待
        }
    }.collect { value ->
        println("[Consumer] 开始处理 $value (${System.currentTimeMillis()})")
        delay(500)           // 模拟消费者处理较慢
        println("[Consumer] 处理完毕 $value")
    }
}

[Producer] 准备发射 1 (1700000000000)
[Consumer] 开始处理 1 (1700000000001)
[Consumer] 处理完毕 1
[Producer] 准备发射 2 (1700000000502)    ← 等消费者处理完才发射下一个
[Consumer] 开始处理 2 (1700000000502)
[Consumer] 处理完毕 2
[Producer] 准备发射 3 (1700000001003)
[Consumer] 开始处理 3 (1700000001003)
[Consumer] 处理完毕 3

// RxJava —— 需要手动管理 Disposable 和线程调度
userApi.getUsers()                              // 返回 Observable<User>
    .subscribeOn(Schedulers.io())               // 上游在 IO 线程
    .observeOn(AndroidSchedulers.mainThread())  // 下游切到主线程
    .filter(user -> user.isActive())            // 过滤活跃用户
    .map(user -> user.getName())                // 映射为用户名
    .subscribe(                                 // 订阅
        name -> updateUI(name),                 // onNext：更新 UI
        error -> showError(error),              // onError：处理错误
        () -> Log.d("TAG", "Complete")          // onComplete：流结束
    );
// ⚠️ 别忘了在 onDestroy 中 dispose，否则内存泄漏！

// Kotlin Flow —— 协程自动管理生命周期，代码更简洁
viewModelScope.launch {                         // 结构化并发：ViewModel 销毁时自动取消
    userApi.getUsers()                          // 返回 Flow<User>
        .flowOn(Dispatchers.IO)                 // 上游在 IO 调度器执行
        .filter { user -> user.isActive }       // 过滤活跃用户
        .map { user -> user.name }              // 映射为用户名
        .catch { error -> showError(error) }    // 捕获上游异常（替代 onError）
        .collect { name ->                      // 收集（替代 subscribe + onNext）
            updateUI(name)                      // collect 默认在 launch 所在的调度器执行
        }
}
// ✅ 无需手动 dispose！viewModelScope 会在 ViewModel 清除时自动取消协程

// ==================== Sequence（同步、不可挂起） ====================
val mySequence: Sequence<Int> = sequence {
    // yield() 不是 suspend 函数（它是 SequenceScope 的受限挂起）
    yield(1)            // 产出第一个元素
    Thread.sleep(1000)  // ⚠️ 只能用阻塞方式模拟"等待"
    yield(2)            // 产出第二个元素
}
 
// ==================== Flow（异步、可挂起） ==========================
val myFlow: Flow<Int> = flow {
    emit(1)             // 产出第一个元素
    delay(1000)         // ✅ 真正的非阻塞挂起，不占用线程
    emit(2)             // 产出第二个元素
}

// Kotlin 标准库源码（简化）
public fun <T> sequence(
    block: suspend SequenceScope<T>.() -> Unit  // 看起来也是 suspend？
): Sequence<T>

@RestrictsSuspension                     // 🔑 关键注解
public abstract class SequenceScope<in T> internal constructor() {
    public abstract suspend fun yield(value: T)        // 挂起，但受限
    public abstract suspend fun yieldAll(iterator: Iterator<T>)
}

val brokenSequence = sequence {
    yield(1)
    delay(1000)   // ❌ 编译错误！Restricted suspending functions can only
                  //    invoke member or extension suspending functions on their
                  //    restricted coroutine scope
    yield(2)
}

// kotlinx.coroutines 源码（简化）
public fun <T> flow(
    block: suspend FlowCollector<T>.() -> Unit  // 完整的 suspend，无限制
): Flow<T>

// 一个模拟"每秒从传感器读取温度"的 Flow
fun temperatureFlow(): Flow<Double> = flow {
    while (true) {                         // 无限循环，持续产出数据
        val temp = readSensorAsync()       // ✅ 调用 suspend 网络/硬件函数
        emit(temp)                         // 产出当前温度值
        delay(1000)                        // ✅ 非阻塞等待 1 秒
    }
}
 
// 模拟的挂起函数
suspend fun readSensorAsync(): Double {
    delay(200)                             // 模拟 I/O 耗时
    return Math.random() * 40             // 返回 0~40 之间的随机温度
}

// ======================== Sequence 方案（阻塞） ========================
fun loadUsersSequence(): Sequence<User> = sequence {
    val ids = listOf(1, 2, 3)              // 要加载的用户 ID
    for (id in ids) {
        // Thread.sleep 会阻塞调用线程（如主线程），非常危险！
        Thread.sleep(1000)                 // ⚠️ 模拟耗时，但这会冻结 UI
        val user = User(id, "User_$id")   // 构建用户对象
        yield(user)                        // 产出用户
    }
}
 
// 调用方 —— 在主线程逐个迭代，每次都会卡顿 1 秒
fun main() {
    loadUsersSequence().forEach { user ->
        println(user)                      // 每隔 1 秒才打印一次
    }
}

// ======================== Flow 方案（非阻塞） =========================
fun loadUsersFlow(): Flow<User> = flow {
    val ids = listOf(1, 2, 3)              // 要加载的用户 ID
    for (id in ids) {
        delay(1000)                        // ✅ 非阻塞挂起，不占用线程
        val user = fetchUser(id)           // ✅ 可以调用真正的 suspend 网络请求
        emit(user)                         // 产出用户
    }
}
 
// 真正的挂起函数：模拟网络请求
suspend fun fetchUser(id: Int): User {
    delay(500)                             // 模拟网络延迟
    return User(id, "User_$id")
}
 
// 调用方 —— 必须在协程中 collect
fun main() = runBlocking {
    loadUsersFlow().collect { user ->
        println(user)                      // 每隔 1.5 秒打印一次（1s + 0.5s）
    }
}

fun main() {
    // 这段代码在 main 线程执行
    val seq = sequence {
        println("产出线程: ${Thread.currentThread().name}")   // main
        yield(1)
        println("产出线程: ${Thread.currentThread().name}")   // main
        yield(2)
    }
 
    // forEach 也在 main 线程
    seq.forEach { value ->
        println("消费线程: ${Thread.currentThread().name}")   // main
        println("值: $value")
    }
}

产出线程: main
消费线程: main
值: 1
产出线程: main
消费线程: main
值: 2

fun userFlow(): Flow<User> = flow {
    // 这里的代码将由 flowOn 指定的调度器执行
    println("Emit 线程: ${Thread.currentThread().name}")   // DefaultDispatcher-worker-X
    val user = fetchUserFromNetwork()   // 耗时网络请求
    emit(user)                          // 产出用户数据
}
.flowOn(Dispatchers.IO)                 // 🔑 上游切换到 IO 线程池
 
// Android Activity 中的调用
lifecycleScope.launch {                 // 默认在 Main 线程
    userFlow().collect { user ->
        // collect 在 Main 线程执行，可以安全更新 UI
        println("Collect 线程: ${Thread.currentThread().name}")  // main
        textView.text = user.name
    }
}

flow {
    // ② 这里在 IO 线程执行
    emit(heavyComputation())
}
.map { value ->
    // ② 这里也在 IO 线程（在 flowOn 上游）
    transform(value)
}
.flowOn(Dispatchers.IO)        // 🔑 分界线：以上所有都切到 IO
.filter { value ->
    // ③ 这里在 collect 的上下文（如 Main）执行
    value > threshold
}
.collect { value ->
    // ③ 这里也在 collect 的上下文执行
    updateUI(value)
}

flow {
    // ① 在 Default 线程（被最近的 flowOn 决定）
    emit(parseLargeJson())
}
.flowOn(Dispatchers.Default)    // 🔑 第一个分界线
.map { data ->
    // ② 在 IO 线程（被下面的 flowOn 决定）
    saveToDatabase(data)
    data
}
.flowOn(Dispatchers.IO)         // 🔑 第二个分界线
.collect { data ->
    // ③ 在 Main 线程（collect 的协程上下文）
    showData(data)
}

// ❌ 错误用法！运行时会抛出 IllegalStateException
flow {
    withContext(Dispatchers.IO) {    // 🚫 禁止！
        emit(fetchData())
    }
}

// ✅ 正确用法
flow {
    emit(fetchData())               // 在 flow 内部正常 emit
}
.flowOn(Dispatchers.IO)             // 通过 flowOn 切换上下文

// ViewModel 中
class UserViewModel : ViewModel() {
 
    // 定义一个 Flow：从网络加载 → 存入数据库 → 返回结果
    val userData: Flow<UiState<User>> = flow {
        emit(UiState.Loading)                          // 先发射 Loading 状态
 
        val user = apiService.fetchUser()              // ✅ suspend 网络请求
        database.insertUser(user)                      // ✅ suspend 数据库操作
 
        emit(UiState.Success(user))                    // 发射 Success 状态
    }
    .catch { e ->                                      // 捕获上游异常
        emit(UiState.Error(e.message ?: "Unknown"))    // 发射 Error 状态
    }
    .flowOn(Dispatchers.IO)                            // 🔑 上游全部在 IO 线程执行
 
    // 如果用 Sequence？
    // ❌ 无法调用 suspend 函数（apiService.fetchUser 是 suspend 的）
    // ❌ 无法切换到 IO 线程（Sequence 没有 flowOn）
    // ❌ 在主线程执行网络请求 → 直接 Crash (NetworkOnMainThreadException)
}
 
// Activity / Fragment 中
lifecycleScope.launch {                                // 在 Main 线程启动
    viewModel.userData.collect { state ->              // collect 在 Main 线程
        when (state) {
            is UiState.Loading -> showProgress()       // 显示加载动画
            is UiState.Success -> showUser(state.data) // 显示用户数据
            is UiState.Error -> showError(state.msg)   // 显示错误信息
        }
    }
}

fun main() = runBlocking {
    flow {
        println("A: ${Thread.currentThread().name}")
        emit(1)
    }
    .map {
        println("B: ${Thread.currentThread().name}")
        it * 10
    }
    .flowOn(Dispatchers.Default)
    .filter {
        println("C: ${Thread.currentThread().name}")
        it > 5
    }
    .collect {
        println("D: ${Thread.currentThread().name}")
    }
}

// 导入 flow 构建器和相关函数
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.delay
 
// 使用 flow 构建器创建一个 Flow<Int>
// 注意：此时 Lambda 内部的代码 **不会** 执行（冷流特性）
val numberFlow: Flow<Int> = flow {
    // --- 以下代码只有在 collect 时才会运行 ---
    println("Flow started")          // 当收集开始时，才打印这行
    emit(1)                          // 向下游发射第一个值
    delay(500)                       // 挂起 500ms，模拟异步耗时操作
    emit(2)                          // 发射第二个值
    delay(500)                       // 再次挂起
    emit(3)                          // 发射第三个值
    println("Flow completed")        // 所有值发射完毕
}

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.delay
 
// 模拟从分页 API 拉取数据的 Flow
// 每次请求一页，直到没有更多数据为止
fun fetchPages(): Flow<List<String>> = flow {
    var page = 1                     // 初始页码
    var hasMore = true               // 是否还有下一页
 
    while (hasMore) {                // 循环拉取，直到无更多数据
        val data = fakeApiCall(page) // 调用挂起函数，模拟网络请求
        emit(data.items)             // 将本页数据发射给下游
        hasMore = data.hasNextPage   // 更新分页标志
        page++                       // 页码自增
    }
    // 循环结束，Flow 自然完成（相当于 onComplete）
}
 
// 模拟 API 响应的数据类
data class PageResponse(
    val items: List<String>,         // 当前页的数据列表
    val hasNextPage: Boolean         // 是否存在下一页
)
 
// 模拟挂起的网络请求函数
suspend fun fakeApiCall(page: Int): PageResponse {
    delay(1000)                      // 模拟 1 秒网络延迟
    return when (page) {             // 根据页码返回不同的模拟数据
        1 -> PageResponse(listOf("A", "B", "C"), hasNextPage = true)
        2 -> PageResponse(listOf("D", "E", "F"), hasNextPage = true)
        3 -> PageResponse(listOf("G", "H"), hasNextPage = false)  // 最后一页
        else -> PageResponse(emptyList(), hasNextPage = false)
    }
}

import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking
 
val repeatable: kotlinx.coroutines.flow.Flow<Int> = flow {
    println(">>> Flow lambda 执行了一次")  // 每次 collect 都会打印
    emit(42)                                // 每次 collect 都会发射 42
}
 
fun main() = runBlocking {
    println("第一次收集:")
    repeatable.collect { value ->            // 第一次触发 flow lambda
        println("收到: $value")
    }
 
    println("\n第二次收集:")
    repeatable.collect { value ->            // 第二次触发 flow lambda（重新从头执行）
        println("收到: $value")
    }
}
// 输出:
// 第一次收集:
// >>> Flow lambda 执行了一次
// 收到: 42
//
// 第二次收集:
// >>> Flow lambda 执行了一次
// 收到: 42

import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch
import kotlinx.coroutines.coroutineScope
 
// ❌ 错误示范：在 flow 内部启动并发协程来 emit
val badFlow = flow {
    coroutineScope {
        launch { emit(1) }           // 协程 A 尝试 emit
        launch { emit(2) }           // 协程 B 同时 emit → 💥 IllegalStateException
    }
}
 
// ✅ 正确做法：顺序 emit
val goodFlow = flow {
    emit(1)                           // 先发射 1
    emit(2)                           // 再发射 2（顺序执行，安全）
}

import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.Flow
 
// 创建一个包含三个 Int 值的 Flow
// 类似于 listOf(1, 2, 3) 之于 List
val simpleFlow: Flow<Int> = flowOf(1, 2, 3)
 
// 创建只有单个元素的 Flow
val singleFlow: Flow<String> = flowOf("Hello")
 
// 创建空 Flow（不发射任何值，直接完成）
// 等价于 emptyFlow<String>()
val emptyFlow: Flow<String> = flowOf()

// flowOf 的简化源码实现
// 它本质上就是对 flow { emit() } 的封装
public fun <T> flowOf(vararg elements: T): Flow<T> = flow {
    for (element in elements) {      // 遍历传入的所有参数
        emit(element)                // 逐个 emit 给下游
    }
}

flow {
    emit(1)                          // 发射 1
    emit(2)                          // 发射 2
    emit(3)                          // 发射 3
}

import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.Flow
 
// 场景 1：单元测试中构造 Mock 数据源
// 在测试 ViewModel 时，用 flowOf 替代真实的 Repository Flow
fun createMockUserFlow(): Flow<String> {
    return flowOf("Alice", "Bob", "Charlie") // 快速构造测试数据
}
 
// 场景 2：提供默认值 / 兜底数据
// 当缓存为空时返回一组默认配置
fun getDefaultConfig(): Flow<Map<String, String>> {
    return flowOf(                   // 包装一个固定的默认配置
        mapOf(
            "theme" to "light",      // 默认浅色主题
            "language" to "zh-CN"    // 默认中文
        )
    )
}
 
// 场景 3：与操作符组合，构造简单的管道
// flowOf 产出原始值 → map 做变换 → collect 消费
fun demo() {
    val transformed = flowOf(1, 2, 3, 4, 5)  // 创建源 Flow
        // .map { it * 10 }                   // 可链式调用操作符
    // transformed.collect { ... }            // 收集结果
}

import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.Flow
 
// ① List → Flow
val listFlow: Flow<Int> = listOf(1, 2, 3).asFlow()
 
// ② IntRange → Flow（Range 实现了 Iterable 接口）
val rangeFlow: Flow<Int> = (1..10).asFlow()
 
// ③ Sequence → Flow
val seqFlow: Flow<Int> = sequenceOf(10, 20, 30).asFlow()
 
// ④ Array → Flow
val arrayFlow: Flow<String> = arrayOf("X", "Y", "Z").asFlow()
 
// ⑤ 普通函数 → Flow（发射该函数的返回值，仅一个元素）
val funcFlow: Flow<Long> = ::currentTimeMillis.asFlow()
 
// ⑥ 挂起函数 → Flow（发射挂起函数的返回值，仅一个元素）
val suspendFuncFlow: Flow<String> = (suspend { fetchUserName() }).asFlow()

// Iterable<T>.asFlow() 的简化源码
public fun <T> Iterable<T>.asFlow(): Flow<T> = flow {
    forEach { value ->               // 遍历 Iterable 的每个元素
        emit(value)                  // 逐个 emit
    }
}

// suspend () -> T 的 asFlow 简化源码
public fun <T> (suspend () -> T).asFlow(): Flow<T> = flow {
    emit(invoke())                   // 调用挂起函数，将返回值作为唯一元素 emit
}

import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
 
fun main() = runBlocking {
    // Sequence 版本：
    // map 内部不能调用 delay() 等挂起函数
    // Sequence 的操作符不是 suspend 的
    val seq = sequenceOf(1, 2, 3)
        .map { it * 2 }             // 普通 lambda，不支持 suspend
 
    // Flow 版本：
    // 通过 asFlow() 桥接后，进入 Flow 的世界
    // 所有中间操作符都支持挂起
    sequenceOf(1, 2, 3)
        .asFlow()                    // Sequence → Flow
        .map { value ->              // Flow 的 map 接受 suspend lambda
            delay(100)               // ✅ 可以挂起！Sequence 做不到
            value * 2                // 变换逻辑
        }
        .collect { result ->         // 终端操作符，触发整个流水线
            println(result)          // 输出: 2, 4, 6（每个间隔 100ms）
        }
}

import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
 
// 假设从数据库拿到了一批用户 ID
val userIds: List<Int> = listOf(101, 102, 103, 104)
 
// 模拟挂起的数据库查询
suspend fun queryUserById(id: Int): String {
    // delay(50)                     // 实际场景中会有 IO 延迟
    return "User-$id"                // 返回模拟的用户名
}
 
fun main() = runBlocking {
    val userNames: List<String> = userIds
        .asFlow()                    // List<Int> → Flow<Int>
        .map { id ->                 // 对每个 ID 执行异步查询
            queryUserById(id)        // 调用挂起函数（Sequence 无法做到）
        }
        .toList()                    // 收集所有结果为 List
 
    println(userNames)               // 输出: [User-101, User-102, User-103, User-104]
}

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
 
fun main() = runBlocking {
    val myFlow = flow {
        println("A")
        emit(1)
        println("B")
        emit(2)
        println("C")
    }
 
    println("Before first collect")
    myFlow.collect { print("$it ") }
    println()
 
    println("Before second collect")
    myFlow.collect { print("$it ") }
    println()
}

// Flow 接口中唯一的抽象方法
// collect 是 suspend 函数，必须在协程中调用
public suspend fun collect(collector: FlowCollector<T>)

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun simpleFlow(): Flow<Int> = flow {
    // 模拟三次异步数据发射
    for (i in 1..3) {
        delay(100)        // 模拟异步耗时操作（如网络请求）
        emit(i)           // 向下游发射一个值
    }
}
 
fun main() = runBlocking {
    // collect 是 suspend 函数，必须在协程作用域中调用
    simpleFlow().collect { value ->   // lambda 即 FlowCollector.emit 的简写
        println("Received: $value")   // 每收到一个值就执行这里的逻辑
    }
    // 输出:
    // Received: 1
    // Received: 2
    // Received: 3
}

// FlowCollector 是一个函数式接口（fun interface）
public fun interface FlowCollector<in T> {
    // 这个 emit 就是 collect lambda 里"接收值"的动作
    public suspend fun emit(value: T)
}

// 这两种写法完全等价
// 写法 1: lambda 简写（推荐）
myFlow.collect { value ->
    println(value)          // lambda 体就是 FlowCollector.emit 的实现
}
 
// 写法 2: 显式实现 FlowCollector 接口
myFlow.collect(object : FlowCollector<Int> {
    override suspend fun emit(value: Int) {
        println(value)      // 与写法 1 完全等价
    }
})

fun main() = runBlocking {
    flowOf(1, 2, 3).collect { value ->
        delay(500)                     // 在 collect 中挂起是完全合法的
        println("Slow consume: $value") // 模拟慢消费者
    }
    // 每个值之间间隔 500ms 输出
}

val countFlow = flow {
    println("Flow started!")   // 每次 collect 都会打印
    for (i in 1..3) {
        emit(i)                // 每次 collect 都会重新发射 1, 2, 3
    }
}
 
fun main() = runBlocking {
    println("=== 第一次收集 ===")
    countFlow.collect { println(it) }   // 打印 Flow started! 然后 1 2 3
 
    println("=== 第二次收集 ===")
    countFlow.collect { println(it) }   // 再次打印 Flow started! 然后 1 2 3
}
// 输出:
// === 第一次收集 ===
// Flow started!
// 1
// 2
// 3
// === 第二次收集 ===
// Flow started!
// 1
// 2
// 3

fun main() = runBlocking {
    val flow = flow {
        emit(1)
        delay(1000)          // 模拟耗时
        emit(2)
    }
 
    flow.collect { println(it) }    // 整个 collect 结束前，下面的代码不会执行
    println("Done!")                 // 在 emit(2) 被消费之后才打印
    // 输出:
    // 1
    // (等待 1 秒)
    // 2
    // Done!
}

fun main() = runBlocking {
    val flow = flowOf(1, 2, 3)
 
    launch {                           // 在独立协程中收集
        flow.collect { println(it) }   // 不会阻塞外部 runBlocking 的后续代码
    }
 
    println("This prints immediately") // 可能先于 flow 的输出打印
}

fun main() = runBlocking {
    val infiniteFlow = flow {
        var i = 0
        while (true) {                // 无限流
            emit(i++)                 // 不断发射递增的整数
            delay(100)                // 每 100ms 发射一个
        }
    }
 
    // withTimeoutOrNull 在超时后会取消内部协程
    val result = withTimeoutOrNull(350) {
        infiniteFlow.collect { println(it) }  // 超时后 collect 自动取消
    }
    println("Result: $result")        // 超时返回 null
    // 输出:
    // 0
    // 1
    // 2
    // Result: null
}

// toList 的签名（简化版）
public suspend fun <T> Flow<T>.toList(): List<T>
 
// toSet 的签名（简化版）
public suspend fun <T> Flow<T>.toSet(): Set<T>

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    // 创建一个简单的 Flow
    val numberFlow: Flow<Int> = flowOf(1, 2, 3, 4, 5)
 
    // toList(): 将所有发射值收集到一个 List
    val list: List<Int> = numberFlow.toList()   // 挂起直到 Flow 完成
    println("List: $list")                       // List: [1, 2, 3, 4, 5]
 
    // toSet(): 将所有发射值收集到一个 Set（自动去重）
    val dupFlow = flowOf(1, 2, 2, 3, 3, 3)
    val set: Set<Int> = dupFlow.toSet()          // 重复元素会被去除
    println("Set: $set")                          // Set: [1, 2, 3]
}

// toList 的等价手写实现（仅供理解，实际请直接用 toList）
suspend fun <T> Flow<T>.myToList(): List<T> {
    val result = mutableListOf<T>()   // 创建一个可变列表作为容器
    collect { value ->                 // 通过 collect 逐个收集值
        result.add(value)             // 每收到一个值就添加到列表中
    }
    return result                      // Flow 完成后返回完整列表
}

fun main() = runBlocking {
    // ⚠️ 危险: 无限 Flow 调用 toList 永远不会返回！
    // val infiniteList = flow {
    //     var i = 0
    //     while (true) { emit(i++); delay(100) }
    // }.toList()   // 永远挂起，内存无限增长!
 
    // ✅ 正确做法: 先用 take 截取有限个元素
    val safeList = flow {
        var i = 0
        while (true) {                // 无限流
            emit(i++)                 // 持续发射
            delay(100)
        }
    }.take(5)                          // 只取前 5 个，之后自动取消上游
     .toList()                         // 安全: Flow 现在是有限的
    println("Safe list: $safeList")    // Safe list: [0, 1, 2, 3, 4]
}

fun main() = runBlocking {
    val existing = mutableListOf(100, 200)     // 已有数据
    val flow = flowOf(1, 2, 3)
 
    // 将 Flow 的值追加到已有列表中
    val combined = flow.toList(existing)       // existing 会被修改并返回
    println(combined)                           // [100, 200, 1, 2, 3]
    println(existing === combined)              // true，是同一个对象
}

// 两种重载形式
public suspend fun <T> Flow<T>.first(): T                          // 取第一个值
public suspend fun <T> Flow<T>.first(predicate: (T) -> Boolean): T // 取满足条件的第一个值

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    val flow = flowOf(10, 20, 30, 40)
 
    // 获取第一个元素
    val firstValue = flow.first()              // 拿到 10 后立刻停止收集
    println("First: $firstValue")              // First: 10
 
    // 获取满足条件的第一个元素
    val firstEven = flow.first { it > 15 }     // 跳过 10，拿到 20 后停止
    println("First > 15: $firstEven")          // First > 15: 20
}

fun main() = runBlocking {
    val heavyFlow = flow {
        println("Emitting 1")      // 会执行
        emit(1)
        println("Emitting 2")      // 不会执行！因为 first() 已经取消了 Flow
        emit(2)
        println("Emitting 3")      // 同样不会执行
        emit(3)
    }
 
    val result = heavyFlow.first()  // 收到 1 后取消 Flow
    println("Got: $result")
    // 输出:
    // Emitting 1
    // Got: 1
}

// first() 的取消执行流程（概念模型）
 
// 上游 Flow:     emit(1) ──→ emit(2) ──→ emit(3)
//                  │             ✗            ✗
//                  ▼
// first():      收到 1 ──→ 抛出 CancellationException 取消上游
//                  │
//                  ▼
// 返回值:        result = 1

fun main() = runBlocking {
    val emptyFlow = emptyFlow<Int>()           // 创建一个空 Flow
 
    // first() 对空 Flow 会抛异常
    // val value = emptyFlow.first()           // ❌ 抛出 NoSuchElementException
 
    // firstOrNull() 安全版本，空 Flow 返回 null
    val safeValue = emptyFlow.firstOrNull()    // ✅ 返回 null
    println("Safe: $safeValue")                // Safe: null
 
    // 带条件的 firstOrNull
    val flow = flowOf(1, 3, 5)
    val even = flow.firstOrNull { it % 2 == 0 }  // 没有偶数，返回 null
    println("Even: $even")                         // Even: null
}

// single 的签名
public suspend fun <T> Flow<T>.single(): T

fun main() = runBlocking {
    // ✅ 正好一个值 → 正常返回
    val one = flowOf(42).single()
    println("Single: $one")               // Single: 42
 
    // ❌ 空 Flow → NoSuchElementException
    try {
        emptyFlow<Int>().single()
    } catch (e: NoSuchElementException) {
        println("Empty: ${e.message}")    // Empty: Expected at least one element
    }
 
    // ❌ 多个值 → IllegalStateException
    try {
        flowOf(1, 2).single()
    } catch (e: IllegalStateException) {
        println("Too many: ${e.message}") // Too many: Expected only one element
    }
}

fun main() = runBlocking {
    // 空 Flow → null
    val a = emptyFlow<Int>().singleOrNull()
    println(a)  // null
 
    // 恰好一个值 → 返回值
    val b = flowOf(99).singleOrNull()
    println(b)  // 99
 
    // 多个值 → null
    val c = flowOf(1, 2).singleOrNull()
    println(c)  // null
}

// Room DAO 示例（伪代码）
@Query("SELECT * FROM user WHERE id = :userId")
fun getUserById(userId: String): Flow<User>     // Room 返回 Flow
 
// 在 ViewModel 中使用
suspend fun loadUser(id: String): User {
    return userDao.getUserById(id)
        .single()    // 业务断言: 按主键查应该恰好只有 1 条
}

fun main() = runBlocking {
    val flow = flow {
        println("Start")
        emit(1)
        println("After 1")
        emit(2)
        println("After 2")
        emit(3)
        println("After 3")
    }
 
    val result = flow.first { it > 1 }
    println("Result: $result")
}

// Flow<T> 经过 map 后变为 Flow<R>
// transform 是一个挂起函数（suspend），意味着你可以在 map 中调用其他挂起函数
public inline fun <T, R> Flow<T>.map(
    crossinline transform: suspend (value: T) -> R  // 接受 T，返回 R
): Flow<R>

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    // 创建一个发射 1, 2, 3 的简单 Flow
    val numbersFlow: Flow<Int> = flowOf(1, 2, 3)
 
    // 使用 map 将每个数字乘以 10
    numbersFlow
        .map { value ->          // value 的类型是 Int
            println("  map 正在处理: $value")  // 观察处理顺序
            value * 10           // 将每个元素乘以 10，返回新值
        }
        .collect { result ->     // result 的类型仍然是 Int（因为 Int * Int = Int）
            println("收集到: $result")
        }
}

  map 正在处理: 1
收集到: 10
  map 正在处理: 2
收集到: 20
  map 正在处理: 3
收集到: 30

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
// 模拟一个挂起的网络请求：根据用户 ID 获取用户名
suspend fun fetchUserName(userId: Int): String {
    delay(100)                              // 模拟 100ms 网络延迟
    return "User_$userId"                   // 返回模拟的用户名
}
 
fun main() = runBlocking {
    val userIds = listOf(101, 102, 103)     // 三个用户 ID
 
    userIds
        .asFlow()                           // List<Int> 转换为 Flow<Int>
        .map { id ->                        // 对每个 ID 进行异步转换
            fetchUserName(id)               // 调用挂起函数，Flow 会自动挂起等待结果
        }
        .collect { name ->                  // 收集转换后的用户名
            println("获取到用户名: $name")
        }
}

获取到用户名: User_101
获取到用户名: User_102
获取到用户名: User_103

// 输入 Flow<T>，输出仍然是 Flow<T>（类型不变，数量可能减少）
public inline fun <T> Flow<T>.filter(
    crossinline predicate: suspend (value: T) -> Boolean  // 同样是 suspend 函数
): Flow<T>

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    // 创建一个发射 1 到 10 的 Flow
    val numbersFlow: Flow<Int> = (1..10).asFlow()
 
    // 只保留偶数
    numbersFlow
        .filter { value ->               // value 类型是 Int
            val isEven = value % 2 == 0  // 判断是否为偶数
            println("  filter 判断 $value → $isEven")
            isEven                       // 返回 true 则保留，false 则丢弃
        }
        .collect { result ->
            println("收集到: $result")
        }
}

  filter 判断 1 → false
  filter 判断 2 → true
收集到: 2
  filter 判断 3 → false
  filter 判断 4 → true
收集到: 4
  filter 判断 5 → false
  filter 判断 6 → true
收集到: 6
  filter 判断 7 → false
  filter 判断 8 → true
收集到: 8
  filter 判断 9 → false
  filter 判断 10 → true
收集到: 10

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
data class Product(                       // 定义一个商品数据类
    val name: String,                     // 商品名称
    val price: Double,                    // 商品价格
    val inStock: Boolean                  // 是否有库存
)
 
fun main() = runBlocking {
    // 模拟商品列表数据源
    val products = listOf(
        Product("Keyboard", 79.99, true),
        Product("Mouse", 39.99, false),   // 无库存
        Product("Monitor", 299.99, true),
        Product("Webcam", 59.99, true),
        Product("Speaker", 149.99, false) // 无库存
    )
 
    products
        .asFlow()                         // List<Product> → Flow<Product>
        .filter { it.inStock }            // 第一步：只保留有库存的商品
        .filter { it.price < 100.0 }      // 第二步：只保留价格低于 100 的商品
        .map { it.name }                  // 第三步：提取商品名称
        .collect { name ->
            println("符合条件的商品: $name")
        }
}

符合条件的商品: Keyboard
符合条件的商品: Webcam

// FlowCollector<R> 是接收者，提供了 emit(value: R) 函数
public inline fun <T, R> Flow<T>.transform(
    crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
): Flow<R>

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    val numbersFlow: Flow<Int> = (1..5).asFlow()
 
    numbersFlow
        .transform { value ->             // this: FlowCollector<String>，value: Int
            if (value % 2 == 0) {         // 只处理偶数（过滤效果）
                emit("$value 是偶数")      // 第一次 emit：描述性字符串
                emit("$value × 2 = ${value * 2}")  // 第二次 emit：计算结果
            }
            // 奇数不调用 emit，相当于被过滤掉了
        }
        .collect { result ->
            println(result)
        }
}

2 是偶数
2 × 2 = 4
4 是偶数
4 × 2 = 8

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
// 模拟发起网络请求
suspend fun performRequest(id: Int): String {
    delay(200)                                // 模拟网络延迟
    return "Response_$id"                     // 返回响应结果
}
 
fun main() = runBlocking {
    (1..3).asFlow()
        .transform { requestId ->
            emit("⏳ 正在请求 #$requestId...")    // 先发射一个"加载中"的状态
            val response = performRequest(requestId)  // 执行异步请求
            emit("✅ $response")               // 再发射请求结果
        }
        .collect { status ->
            println(status)
        }
}

⏳ 正在请求 #1...
✅ Response_1
⏳ 正在请求 #2...
✅ Response_2
⏳ 正在请求 #3...
✅ Response_3

// 只有前 count 个元素会被传递给下游
public fun <T> Flow<T>.take(count: Int): Flow<T>

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    // 创建一个"无限" Flow（实际上 1..100 足够演示）
    val numbersFlow: Flow<Int> = (1..100).asFlow()
 
    // 只取前 5 个元素
    numbersFlow
        .take(5)                          // 只允许前 5 个元素通过
        .collect { value ->
            println("收集到: $value")
        }
 
    println("--- Flow 已结束 ---")
}

收集到: 1
收集到: 2
收集到: 3
收集到: 4
收集到: 5
--- Flow 已结束 ---

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    flow {
        for (i in 1..10) {
            println("  准备发射 $i")       // 观察哪些元素实际被发射了
            emit(i)                       // 发射元素
            println("  已发射 $i")         // 观察发射后是否继续
        }
    }
    .take(3)                              // 只取前 3 个
    .collect { value ->
        println("收集到: $value")
    }
}

  准备发射 1
  已发射 1
收集到: 1
  准备发射 2
  已发射 2
收集到: 2
  准备发射 3
  已发射 3
收集到: 3

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    flow {
        try {
            for (i in 1..10) {
                emit(i)                   // 逐个发射
            }
        } finally {
            println("🧹 finally 块执行：资源清理")  // take 取消后会触发这里
        }
    }
    .take(2)                              // 只取前 2 个
    .collect { println("收集到: $it") }
}

收集到: 1
收集到: 2
🧹 finally 块执行：资源清理