Flow进阶 ⭐⭐

// 示例：默认情况下 emit 和 collect 运行在同一上下文
fun main() = runBlocking {
    // 这个 flow 的 emit 会继承 collect 所在的上下文（此处为 runBlocking 的主线程）
    val simpleFlow = flow {
        println("Flow 发射线程: ${Thread.currentThread().name}") // 主线程
        emit(1) // 在 collect 的上下文中发射
        emit(2)
    }
 
    // collect 运行在 runBlocking 的主线程上
    simpleFlow.collect { value ->
        println("收集到 $value, 线程: ${Thread.currentThread().name}") // 主线程
    }
}
// 输出：
// Flow 发射线程: main
// 收集到 1, 线程: main
// 收集到 2, 线程: main

fun main() = runBlocking {
    flow {
        // ✅ 这里运行在 Dispatchers.IO（由下方 flowOn 指定）
        println("发射线程: ${Thread.currentThread().name}")
        emit(fetchDataFromNetwork()) // 模拟耗时网络请求
    }
    .map { data ->
        // ✅ 这里也运行在 Dispatchers.IO（处于 flowOn 上游）
        println("map 变换线程: ${Thread.currentThread().name}")
        parseData(data) // 解析数据
    }
    .flowOn(Dispatchers.IO) // 🔑 分水岭：以上所有操作切换到 IO 线程
    .collect { parsed ->
        // ✅ 这里运行在 runBlocking 的主线程（collect 的上下文）
        println("收集线程: ${Thread.currentThread().name}")
        updateUI(parsed) // 更新 UI
    }
}
// 输出（线程名因环境而异）：
// 发射线程: DefaultDispatcher-worker-1
// map 变换线程: DefaultDispatcher-worker-1
// 收集线程: main

fun main() = runBlocking {
    flow {
        // 🟠 运行在 Dispatchers.IO（受第一个 flowOn 控制）
        println("emit 线程: ${Thread.currentThread().name}")
        emit(1)
    }
    .flowOn(Dispatchers.IO) // 第一个 flowOn：控制 flow{} 构建块
    .map { value ->
        // 🔵 运行在 Dispatchers.Default（受第二个 flowOn 控制）
        println("map 线程: ${Thread.currentThread().name}")
        value * 10
    }
    .flowOn(Dispatchers.Default) // 第二个 flowOn：控制上方的 map
    .collect { value ->
        // 🟢 运行在主线程（collect 自身的上下文）
        println("collect 线程: ${Thread.currentThread().name}")
    }
}

// 概念性内存模型（简化）
// ┌────────────────────┐          ┌───────────┐          ┌────────────────────┐
// │    上游协程          │          │  Channel   │          │    下游协程          │
// │  Dispatchers.IO     │ ──emit──▶│ (buffer=64)│ ──recv──▶│  Dispatchers.Main  │
// │  flow { emit(x) }  │          │            │          │  collect { ... }   │
// └────────────────────┘          └───────────┘          └────────────────────┘

// ❌ 错误写法：绝对不要这样做！
fun wrongFlow(): Flow<Int> = flow {
    // 尝试在 withContext 中切换到 IO 线程后发射
    withContext(Dispatchers.IO) {
        emit(1) // 💥 运行时抛出 IllegalStateException!
    }
}
 
fun main() = runBlocking {
    wrongFlow().collect { println(it) }
    // 异常信息：
    // Flow invariant is violated:
    //     Flow was collected in [BlockingCoroutine{Active}@xxx, BlockingEventLoop@xxx],
    //     but emission happened in [DispatchedCoroutine{Active}@xxx, Dispatchers.IO].
    //     Please refer to 'flow' documentation or use 'flowOn' instead.
}

// ✅ 正确写法：用 flowOn 切换上游上下文
fun correctFlow(): Flow<Int> = flow {
    // 这里的代码由 flowOn 切换到 IO 线程执行
    val data = fetchFromNetwork() // 耗时操作
    emit(data)                    // 安全发射，flowOn 内部处理了上下文切换
}.flowOn(Dispatchers.IO)          // 整个 flow{} 块运行在 IO 线程

// ✅ 正确写法：withContext 只包裹计算逻辑，emit 在外部
fun partialSwitchFlow(): Flow<Int> = flow {
    // 在 IO 线程执行耗时计算，但不在 withContext 内 emit
    val result = withContext(Dispatchers.IO) {
        heavyComputation() // 只做计算，返回结果
    }
    // 回到原上下文后安全发射
    emit(result) // ✅ 这是合法的，因为 emit 在 flow{} 的原始上下文中
}

// ✅ 正确写法：channelFlow 允许在任意上下文中发送
fun concurrentFlow(): Flow<Int> = channelFlow {
    // 可以在不同协程/上下文中并发发送数据
    withContext(Dispatchers.IO) {
        send(fetchFromNetwork()) // ✅ channelFlow 中的 send 是线程安全的
    }
    launch(Dispatchers.Default) {
        send(computeLocally())   // ✅ 也可以从其他协程发送
    }
}

fun main() = runBlocking {
    flow {
        withContext(Dispatchers.IO) {
            println("Fetching data...")
            emit(42)
        }
    }
    .flowOn(Dispatchers.Default)
    .collect { println("Got: $it") }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.measureTimeMillis
 
fun simpleFlow(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100) // 模拟上游每 100ms 产出一个值
        emit(i)    // 发射值
        println("Emitted $i @ ${System.currentTimeMillis()}")
    }
}
 
fun main() = runBlocking {
    val time = measureTimeMillis {
        simpleFlow().collect { value ->
            delay(300) // 模拟下游处理耗时 300ms
            println("Collected $value @ ${System.currentTimeMillis()}")
        }
    }
    // 预期总耗时 ≈ 3 × (100 + 300) = 1200ms
    println("Total time: ${time}ms")
}

Emitted 1
Collected 1
Emitted 2
Collected 2
Emitted 3
Collected 3
Total time: ~1200ms

fun main() = runBlocking {
    val time = measureTimeMillis {
        simpleFlow()
            .buffer() // 在 emit 和 collect 之间创建缓冲通道
            .collect { value ->
                delay(300) // 下游依然需要 300ms 处理
                println("Collected $value")
            }
    }
    // 预期总耗时 ≈ 100 + 3 × 300 = 1000ms
    // 第一个值需要 100ms 生产，之后下游连续消费 3 个值各 300ms
    println("Total time: ${time}ms")
}

public fun <T> Flow<T>.buffer(
    capacity: Int = BUFFERED,           // 缓冲区容量
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND // 溢出策略
): Flow<T>

fun main() = runBlocking {
    flow {
        for (i in 1..5) {
            println("Emitting $i")
            emit(i)            // 发射值
            delay(100)         // 上游每 100ms 发射一个
        }
    }
    .buffer(
        capacity = 1,                           // 缓冲区只能存 1 个值
        onBufferOverflow = BufferOverflow.DROP_OLDEST // 满了就丢弃最旧的
    )
    .collect { value ->
        delay(300)             // 下游每 300ms 处理一个
        println("Collected $value")
    }
}

// conflate() 的内部实现等价于：
buffer(capacity = Channel.CONFLATED)
// 也等价于：
buffer(capacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)

fun main() = runBlocking {
    val time = measureTimeMillis {
        simpleFlow()       // 上游：100ms 发射一个，共 3 个值
            .conflate()    // 合并：下游来不及就丢弃旧值
            .collect { value ->
                delay(300) // 下游处理需要 300ms
                println("Collected $value")
            }
    }
    println("Total time: ${time}ms")
}

Collected 1
Collected 3
Total time: ~700ms

// Android 典型场景：搜索框输入
searchQueryFlow
    .conflate()        // 用户打字快于网络请求，丢弃中间状态
    .debounce(300)     // 再加防抖，等用户停止输入 300ms
    .collect { query ->
        // 只对最新的搜索词发起网络请求
        performSearch(query)
    }

fun main() = runBlocking {
    val time = measureTimeMillis {
        simpleFlow()            // 上游：100ms 发射一个，共 3 个值
            .collectLatest { value ->
                println("Start processing $value")
                delay(300)      // 模拟耗时处理（可被取消的挂起点）
                println("Done processing $value")
            }
    }
    println("Total time: ${time}ms")
}

Start processing 1
Start processing 2
Start processing 3
Done processing 3
Total time: ~700ms

flow {
    emit(1)
    delay(100)
    emit(2)
}.collectLatest { value ->
    // ⚠️ 这是一个非挂起的密集循环，cancel 无法中断它
    var sum = 0L
    for (i in 1..1_000_000_000) {
        sum += i  // 没有挂起点，取消不会生效
    }
    println("Done $value, sum=$sum")
}

.collectLatest { value ->
    var sum = 0L
    for (i in 1..1_000_000_000) {
        sum += i
        if (i % 1_000_000 == 0L) {
            currentCoroutineContext().ensureActive() // 手动检查取消状态
        }
    }
    println("Done $value, sum=$sum")
}

// Android ViewModel 典型用法
viewModelScope.launch {
    repository.getItemsFlow()   // Room 返回 Flow<List<Item>>
        .collectLatest { items ->
            // 如果数据快速连续变更，旧的 UI 更新会被取消
            _uiState.value = UiState.Success(items)
        }
}

fun main() = runBlocking {
    flow {
        emit(1)
        delay(50)
        emit(2)
        delay(50)
        emit(3)
    }.collectLatest { value ->
        delay(100)
        println("Done $value")
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    // 定义第一个 Flow：发射数字 1, 2, 3
    val numbers: Flow<Int> = flow {
        for (i in 1..3) {
            delay(300L)          // 每 300ms 发射一个
            emit(i)              // 发射当前数字
            println("Numbers: 发射 $i")
        }
    }
 
    // 定义第二个 Flow：发射字符串 "a", "b", "c", "d"
    val letters: Flow<String> = flow {
        val items = listOf("a", "b", "c", "d")
        for (item in items) {
            delay(500L)          // 每 500ms 发射一个（比 numbers 慢）
            emit(item)           // 发射当前字母
            println("Letters: 发射 $item")
        }
    }
 
    // 使用 zip 将两个 Flow 按序配对
    numbers.zip(letters) { num, letter ->
        // 这个 lambda 接收配对好的两个值
        "$num -> $letter"        // 拼接成字符串作为输出
    }.collect { result ->
        println("Collect: $result")  // 收集并打印结果
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    // Flow A：快速发射数字，每 300ms 一次
    val numbers: Flow<Int> = flow {
        for (i in 1..3) {
            delay(300L)              // 每 300ms
            emit(i)                  // 发射 i
            println("  [A] 发射: $i")
        }
    }
 
    // Flow B：慢速发射字母，每 500ms 一次
    val letters: Flow<String> = flow {
        listOf("a", "b").forEach { ch ->
            delay(500L)              // 每 500ms
            emit(ch)                 // 发射字母
            println("  [B] 发射: $ch")
        }
    }
 
    // combine：任意一方发射新值，都用双方最新值进行组合
    numbers.combine(letters) { num, letter ->
        "$num-$letter"               // 用最新值拼接
    }.collect { result ->
        println("Collect: $result")  // 打印组合结果
    }
}

// 一个典型的 Android ViewModel 中的用法
val uiState: StateFlow<UiState> = combine(
    userRepo.getUserFlow(),         // 用户信息 Flow
    settingsRepo.getThemeFlow(),    // 主题设置 Flow
    networkMonitor.isOnline         // 网络状态 Flow
) { user, theme, isOnline ->
    // 任何一个源变化都会触发这里重新计算
    UiState(user = user, theme = theme, isOnline = isOnline)
}.stateIn(
    scope = viewModelScope,         // 在 ViewModel 作用域中共享
    started = SharingStarted.WhileSubscribed(5000),
    initialValue = UiState.Loading  // 初始状态
)

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
// 模拟根据 ID 请求详情的函数，返回一个 Flow
fun requestDetail(id: Int): Flow<String> = flow {
    delay(300L)                              // 模拟网络延迟
    emit("$id: 基本信息")                     // 发射第一条结果
    delay(300L)                              // 继续加载
    emit("$id: 详细数据")                     // 发射第二条结果
}
 
@OptIn(FlowPreview::class)                  // flatMapConcat 需要 OptIn
fun main() = runBlocking {
    val startTime = System.currentTimeMillis()
 
    // 上游 Flow 发射 3 个 ID
    flowOf(1, 2, 3)
        .flatMapConcat { id ->               // 对每个 ID 串行展开为子 Flow
            println("[${elapsedTime(startTime)}] 开始处理 ID=$id")
            requestDetail(id)                // 返回该 ID 对应的子 Flow
        }
        .collect { value ->                  // 收集展平后的所有结果
            println("[${elapsedTime(startTime)}] 收到: $value")
        }
}
 
// 辅助函数：计算经过的毫秒数
fun elapsedTime(start: Long) = "${System.currentTimeMillis() - start}ms"

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun requestDetail(id: Int): Flow<String> = flow {
    delay(300L)                                  // 模拟网络延迟
    emit("$id: 基本信息")                         // 第一条数据
    delay(300L)                                  // 继续加载
    emit("$id: 详细数据")                         // 第二条数据
}
 
@OptIn(FlowPreview::class)
fun main() = runBlocking {
    val startTime = System.currentTimeMillis()
 
    flowOf(1, 2, 3)
        .flatMapMerge(concurrency = 3) { id ->   // 最多 3 个子 Flow 并行
            println("[${elapsedTime(startTime)}] 开始处理 ID=$id")
            requestDetail(id)                     // 返回子 Flow（并发执行）
        }
        .collect { value ->
            println("[${elapsedTime(startTime)}] 收到: $value")
        }
}
 
fun elapsedTime(start: Long) = "${System.currentTimeMillis() - start}ms"

// 限制最多 2 个子 Flow 同时运行
.flatMapMerge(concurrency = 2) { id ->
    requestDetail(id)
}
// 此时 ID=1 和 ID=2 并行，ID=3 等 ID=1 或 ID=2 完成后才开始

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun requestDetail(id: Int): Flow<String> = flow {
    delay(300L)                                      // 模拟网络延迟
    emit("$id: 基本信息")                             // 发射第一条
    delay(300L)                                      // 继续加载
    emit("$id: 详细数据")                             // 发射第二条
}
 
@OptIn(ExperimentalCoroutinesApi::class)             // flatMapLatest 需要 OptIn
fun main() = runBlocking {
    val startTime = System.currentTimeMillis()
 
    flow {
        emit(1)                                      // 发射 1
        delay(200L)                                  // 200ms 后发射下一个
        emit(2)                                      // 发射 2 → 取消 1 的子 Flow
        delay(200L)                                  // 200ms 后发射下一个
        emit(3)                                      // 发射 3 → 取消 2 的子 Flow
    }
    .flatMapLatest { id ->                           // 每次新值到来，取消旧的子 Flow
        println("[${elapsedTime(startTime)}] 开始处理 ID=$id")
        requestDetail(id)                            // 返回子 Flow
    }
    .collect { value ->
        println("[${elapsedTime(startTime)}] 收到: $value")
    }
}
 
fun elapsedTime(start: Long) = "${System.currentTimeMillis() - start}ms"

// 搜索功能的经典写法
searchQueryFlow                              // 用户输入的搜索关键词 Flow
    .debounce(300L)                          // 防抖：300ms 内没有新输入才发射
    .distinctUntilChanged()                  // 去重：和上次一样的关键词不重复搜索
    .flatMapLatest { query ->               // 最新展开：只执行最新的搜索
        if (query.isEmpty()) {
            flowOf(emptyList())              // 空查询返回空列表
        } else {
            searchRepository.search(query)   // 执行真正的搜索请求
        }
    }
    .collect { results ->
        updateUI(results)                    // 更新 UI
    }

假设你是一个餐厅经理，收到 3 桌客人的订单：
 
flatMapConcat → 做完第 1 桌的所有菜，再做第 2 桌，再做第 3 桌。（串行、安全、慢）
flatMapMerge  → 3 个厨师同时开工，谁先做好谁先上菜。（并行、快、顺序不定）
flatMapLatest → 每来一个新订单就把前一个扔掉，只做最新那桌的菜。（只关心最新）

fun main() = runBlocking {
    val flowA = flowOf(1, 2, 3).onEach { delay(100) }
    val flowB = flowOf("X", "Y").onEach { delay(200) }
 
    flowA.zip(flowB) { a, b -> "$a$b" }
        .collect { println(it) }
}

// catch 是 Flow 的扩展函数（中间操作符）
// action lambda 的 receiver 是 FlowCollector<T>，意味着你可以在里面 emit 值
fun <T> Flow<T>.catch(
    action: suspend FlowCollector<T>.(cause: Throwable) -> Unit // 异常处理 lambda
): Flow<T>

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun simpleFlow(): Flow<Int> = flow {
    emit(1)                      // 正常发射第一个值
    emit(2)                      // 正常发射第二个值
    throw RuntimeException("上游爆炸了！") // 上游主动抛出异常
    emit(3)                      // 这行永远不会执行
}
 
suspend fun main() {
    simpleFlow()                 // 创建上游 Flow
        .catch { e ->            // catch 拦截上游的异常
            println("捕获到异常: ${e.message}") // 打印异常信息
        }
        .collect { value ->      // 下游收集器
            println("收到: $value")  // 消费收到的值
        }
}

suspend fun main() {
    flow {
        emit(1)                          // 上游发射值 1
        emit(2)                          // 上游发射值 2
    }
    .catch { e ->                        // catch 安装在这里
        println("catch 捕获: ${e.message}") // 尝试捕获异常
    }
    .collect { value ->                  // 下游收集器
        check(value <= 1) {              // 当 value=2 时抛出 IllegalStateException
            "下游炸了！value=$value"       // 这个异常发生在 catch 的下游
        }
        println("收到: $value")
    }
}

suspend fun main() {
    flow {
        emit(1)                              // 上游发射值
        emit(2)
    }
    .onEach { value ->                       // 把消费逻辑移到 onEach（属于 catch 的上游）
        check(value <= 1) {                  // 这里抛出的异常现在在 catch 上游了
            "处理失败！value=$value"
        }
        println("收到: $value")
    }
    .catch { e ->                            // 现在 catch 能捕获 onEach 中的异常了
        println("catch 捕获: ${e.message}")
    }
    .collect()                               // 空 collect，仅用于触发 Flow 的收集
}

suspend fun main() {
    flow {
        emit("从网络加载数据...")             // 模拟网络请求
        throw IOException("网络超时")        // 模拟网络失败
    }
    .catch { e ->                           // 捕获上游网络异常
        println("网络异常: ${e.message}")
        emit("从本地缓存加载的降级数据")       // 发射降级数据给下游
    }
    .collect { data ->
        println("UI 显示: $data")           // 下游正常收到降级数据
    }
}

suspend fun main() = coroutineScope {
    val job = launch {
        flow {
            emit(1)
            delay(1000)                     // 模拟耗时操作
            emit(2)                         // 协程被取消后不会执行到这里
        }
        .catch { e ->                       // CancellationException 不会进入这里
            println("catch: ${e.message}")
        }
        .collect {
            println("收到: $it")
        }
    }
    delay(100)                              // 100ms 后取消
    job.cancel()                            // 取消协程
    println("已取消")
}

suspend fun main() {
    flow {
        emit(1)
        throw ArithmeticException("计算异常") // 第一层上游异常
    }
    .catch { e ->                             // 第一个 catch：捕获 flow{} 中的异常
        println("第一层 catch: ${e.message}")
        emit(-1)                              // 发射恢复值
        throw IllegalStateException("恢复失败") // 在 catch 内部又抛出新异常
    }
    .catch { e ->                             // 第二个 catch：捕获第一个 catch 中的异常
        println("第二层 catch: ${e.message}")
        emit(-999)                            // 最终恢复值
    }
    .collect { value ->
        println("收到: $value")
    }
}

// onCompletion 签名
fun <T> Flow<T>.onCompletion(
    action: suspend FlowCollector<T>.(cause: Throwable?) -> Unit // cause 为 null 表示正常完成
): Flow<T>

suspend fun main() {
    // 场景1：正常完成
    println("=== 正常完成 ===")
    flowOf(1, 2, 3)                          // 创建一个有限 Flow
        .onCompletion { cause ->             // Flow 完成时触发
            if (cause == null) {             // cause 为 null → 正常完成
                println("✅ Flow 正常完成")
            }
        }
        .collect { println("收到: $it") }
 
    println()
 
    // 场景2：异常完成
    println("=== 异常完成 ===")
    flow {
        emit(1)
        throw RuntimeException("出错了")     // 上游抛出异常
    }
    .onCompletion { cause ->                 // Flow 完成时触发
        if (cause != null) {                 // cause 不为 null → 异常终止
            println("❌ Flow 异常终止: ${cause.message}")
        }
    }
    .catch { e ->                            // catch 处理异常，防止崩溃
        println("catch 兜底: ${e.message}")
    }
    .collect { println("收到: $it") }
}

suspend fun main() {
    flow {
        emit(1)
        throw RuntimeException("爆炸")       // 上游异常
    }
    .onCompletion { cause ->                 // 能看到异常
        println("onCompletion 看到: ${cause?.message}") // 但不会拦截它
    }
    // 没有 catch！异常会直接传播到 collect，导致崩溃
    .collect { println("收到: $it") }
}

suspend fun main() {
    flowOf(1, 2, 3)
        .onCompletion { cause ->              // 安装在 collect 上游
            if (cause != null) {
                println("onCompletion 看到异常: ${cause.message}")
            } else {
                println("onCompletion: 正常完成")
            }
        }
        .collect { value ->                   // 下游抛出异常
            if (value == 2) {
                throw IllegalStateException("下游不接受2")
            }
            println("收到: $value")
        }
}

suspend fun main() {
    flowOf(1, 2, 3)
        .onCompletion { cause ->
            if (cause == null) {              // 仅在正常完成时发射
                emit(999)                     // 发射一个"结束标记"给下游
            }
        }
        .collect { println("收到: $it") }
}

suspend fun main() {
    flow {
        emit("加载中...")
        throw IOException("网络错误")
    }
    .onCompletion { cause ->                 // 先经过 onCompletion
        println(
            if (cause == null) "✅ 加载完成"
            else "⚠️ 加载结束(异常: ${cause.message})"
        )
        // 实际项目中：隐藏 loading 动画、释放资源等
    }
    .catch { e ->                            // 再经过 catch 消费异常
        println("降级处理: ${e.message}")
        emit("缓存数据")                      // 发射降级数据
    }
    .collect { println("UI: $it") }
}

class UserViewModel(
    private val repo: UserRepository
) : ViewModel() {
 
    // UI 状态，使用 StateFlow 管理
    private val _uiState = MutableStateFlow<UiState>(UiState.Idle)
    val uiState: StateFlow<UiState> = _uiState.asStateFlow()
 
    fun loadUsers() {
        repo.getUsersFlow()                    // 返回 Flow<List<User>>
            .onStart {                         // Flow 开始收集时触发
                _uiState.value = UiState.Loading  // 显示 loading
            }
            .onEach { users ->                 // 处理每一个发射的值
                _uiState.value = UiState.Success(users) // 更新 UI
            }
            .onCompletion { cause ->           // Flow 完成（无论正常/异常）
                // 隐藏 loading、释放资源等
                println("数据流结束, cause=$cause")
            }
            .catch { e ->                      // 捕获上游所有异常
                _uiState.value = UiState.Error(e.message ?: "未知错误")
            }
            .launchIn(viewModelScope)          // 在 ViewModel 作用域中启动收集
    }
}

flow {
    emit(1)
    throw RuntimeException("boom")
}
.onCompletion { cause ->
    println("A: ${cause?.message}")
    emit(2)
}
.catch { e ->
    println("B: ${e.message}")
    emit(3)
}
.collect {
    println("C: $it")
}

// ✅ 正确：用 flowOn 切换上游
flow {
    emit(heavyComputation()) // 在 Default 线程执行
}.flowOn(Dispatchers.Default)
 .collect { value ->        // 在调用者上下文收集
     updateUI(value)
 }
 
// ✅ 正确：用 channelFlow 实现多上下文发射
channelFlow {
    withContext(Dispatchers.IO) {
        send(fetchFromNetwork()) // channelFlow 允许跨上下文 send
    }
}
 
// ❌ 错误：withContext 包裹 emit
flow {
    withContext(Dispatchers.IO) {
        emit(data) // 💥 IllegalStateException
    }
}

// buffer: 所有值都会被处理，但上下游并行
flow.buffer(capacity = 64).collect { process(it) }
 
// conflate: 下游忙时，中间值被最新值覆盖
flow.conflate().collect { renderUI(it) }
 
// collectLatest: 新值到来，旧的收集被 cancel
flow.collectLatest { query ->
    val result = searchAPI(query) // 如果期间有新 query，这里会被取消
    showResults(result)
}

flow {
    emit(riskyOperation())       // 可能抛异常的上游
}
.onCompletion { cause ->         // 先注册完成回调
    if (cause != null) {
        log("Flow 异常终止: $cause")
    } else {
        log("Flow 正常完成")
    }
}
.catch { e ->                    // 再捕获上游异常（包括 onCompletion 之上的）
    emit(fallbackValue)          // 发射兜底值
}
.collect { value ->
    updateUI(value)
}

fun main() = runBlocking {
    flow {
        emit(1)
        throw RuntimeException("Boom")
    }
    .onCompletion { cause ->
        if (cause != null) println("Completed with error")
        println("onCompletion done")
    }
    .catch { e ->
        println("Caught: ${e.message}")
        emit(-1)
    }
    .collect { println("Collect: $it") }
}