协程进阶

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    // 在 runBlocking 作用域内启动两个子协程
    val job1 = launch {
        delay(1000L) // 模拟耗时操作
        println("Job1 完成") // 这行永远不会执行
    }
 
    val job2 = launch {
        delay(500L) // 500ms 后抛出异常
        throw RuntimeException("Job2 出错了!") // 未捕获的异常
    }
 
    // job2 的异常会传播给 runBlocking（父协程）
    // runBlocking 会取消 job1，然后自身也会因异常而终止
    joinAll(job1, job2)
    println("这行也不会执行") // 因为 runBlocking 已经被异常终止
}

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    // async 的异常被"延迟"到 await() 时才暴露
    val deferred = async {
        delay(100L)
        throw ArithmeticException("计算出错") // 异常被存储在 Deferred 中
    }
 
    try {
        val result = deferred.await() // 在这里异常才被抛出
        println("结果: $result")
    } catch (e: ArithmeticException) {
        println("捕获到异常: ${e.message}") // 输出: 捕获到异常: 计算出错
    }
}

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    val scope = CoroutineScope(Job()) // 创建独立作用域
 
    val deferred = scope.async {
        delay(100L)
        throw RuntimeException("async 内部异常")
    }
 
    delay(200L)
    // 即使不调用 await()，scope 也已经因为异常而被取消了
    println("scope 是否活跃: ${scope.isActive}") // 输出: false
}

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    val job = launch {
        val child = launch {
            delay(Long.MAX_VALUE) // 永远挂起
        }
        delay(100L)
        child.cancel() // 取消子协程，抛出 CancellationException
        // 但父协程不会受影响，因为 CancellationException 不会向上传播
        println("父协程仍然存活") // 正常输出
    }
    job.join()
    println("runBlocking 也正常结束") // 正常输出
}

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    // 创建一个异常处理器
    val handler = CoroutineExceptionHandler { coroutineContext, throwable ->
        // coroutineContext: 发生异常的协程的上下文
        // throwable: 捕获到的异常对象
        println("捕获到异常: ${throwable.message}")
        println("协程名称: ${coroutineContext[CoroutineName]?.name}")
    }
 
    // handler 必须安装在根协程或 CoroutineScope 上才有效
    val scope = CoroutineScope(SupervisorJob() + handler + CoroutineName("MyScope"))
 
    scope.launch(CoroutineName("Worker")) {
        throw IllegalStateException("出了点问题")
    }
 
    delay(200L) // 等待异常被处理
    // 输出:
    // 捕获到异常: 出了点问题
    // 协程名称: Worker
}

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    val handler = CoroutineExceptionHandler { _, e ->
        println("Handler 捕获: ${e.message}")
    }
 
    // ❌ 错误用法：handler 安装在子协程上，不会生效
    val job = launch {
        launch(handler) { // 这个 handler 不会被调用!
            throw RuntimeException("子协程异常")
        }
    }
 
    job.join()
    // 异常会直接传播到 runBlocking，导致程序崩溃
}

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    val handler = CoroutineExceptionHandler { _, e ->
        println("Handler 捕获: ${e.message}")
    }
 
    // ✅ 正确用法：handler 安装在根协程（scope 级别）
    val scope = CoroutineScope(SupervisorJob() + handler)
 
    scope.launch {
        launch {
            throw RuntimeException("深层子协程异常")
        }
    }
 
    delay(200L)
    // 输出: Handler 捕获: 深层子协程异常
}

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    val handler = CoroutineExceptionHandler { _, e ->
        println("主异常: ${e.message}")
        // 获取被抑制的异常列表
        e.suppressed.forEach { suppressed ->
            println("被抑制的异常: ${suppressed.message}")
        }
    }
 
    val scope = CoroutineScope(SupervisorJob() + handler)
 
    scope.launch {
        launch {
            try {
                delay(Long.MAX_VALUE)
            } finally {
                // 在取消的 finally 块中再抛异常
                throw ArithmeticException("清理时出错")
            }
        }
        launch {
            delay(10L)
            throw IOException("网络异常") // 第一个异常
        }
    }
 
    delay(300L)
    // 输出:
    // 主异常: 网络异常
    // 被抑制的异常: 清理时出错
}

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    val handler = CoroutineExceptionHandler { _, e ->
        println("Handler: ${e.message}")
    }
 
    // 使用 SupervisorJob 创建作用域
    val supervisor = CoroutineScope(SupervisorJob() + handler)
 
    val job1 = supervisor.launch {
        delay(100L)
        throw RuntimeException("Job1 崩溃了")
        // 异常不会影响 job2
    }
 
    val job2 = supervisor.launch {
        delay(200L)
        println("Job2 正常完成") // 这行会正常执行!
    }
 
    delay(500L)
    // 输出:
    // Handler: Job1 崩溃了
    // Job2 正常完成
}

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    supervisorScope {
        val job1 = launch {
            delay(100L)
            throw RuntimeException("Job1 失败")
            // 异常不会取消 supervisorScope 和 job2
        }
 
        val job2 = launch {
            delay(200L)
            println("Job2 完成了") // 正常输出
        }
 
        // supervisorScope 会等待 job1 和 job2 都结束
    }
    // 注意：如果没有 CoroutineExceptionHandler，
    // job1 的异常仍然会导致程序崩溃
    println("supervisorScope 之后")
}

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    val handler = CoroutineExceptionHandler { _, e ->
        println("Handler: ${e.message}")
    }
 
    val scope = CoroutineScope(SupervisorJob() + handler)
 
    scope.launch { // 这是 supervisor 的直接子协程
        // 但这里面的 launch 用的是普通 Job!
        launch { // 孙子协程
            throw RuntimeException("孙子协程异常")
            // 这个异常会取消父协程（scope.launch 那个），
            // 但不会影响 scope 的其他直接子协程
        }
        launch {
            delay(200L)
            println("这行不会执行") // 被兄弟协程的异常取消了
        }
    }
 
    scope.launch { // 另一个直接子协程
        delay(300L)
        println("我不受影响") // 正常输出，因为 SupervisorJob 隔离了直接子协程
    }
 
    delay(500L)
}

// Android ViewModel 中的典型用法（伪代码示意）
class MyViewModel : ViewModel() {
    // viewModelScope 内部已经使用了 SupervisorJob + Dispatchers.Main.immediate
 
    fun loadDashboard() {
        // 三个请求互不影响（SupervisorJob 的功劳）
        viewModelScope.launch {
            try {
                val avatar = userRepository.loadAvatar() // 具体操作用 try-catch
                _avatarState.value = avatar
            } catch (e: Exception) {
                _avatarState.value = AvatarState.Error(e.message)
            }
        }
 
        viewModelScope.launch {
            try {
                val feeds = feedRepository.loadFeeds()
                _feedState.value = feeds
            } catch (e: Exception) {
                _feedState.value = FeedState.Error(e.message)
            }
        }
 
        viewModelScope.launch {
            try {
                val recommendations = recRepository.loadRecs()
                _recState.value = recommendations
            } catch (e: Exception) {
                _recState.value = RecState.Error(e.message)
            }
        }
    }
}

fun main() = runBlocking {
    val handler = CoroutineExceptionHandler { _, e ->
        println("Caught: ${e.message}")
    }
 
    val scope = CoroutineScope(Job() + handler)
 
    scope.launch {
        launch {
            delay(100)
            throw RuntimeException("Boom")
        }
        launch {
            delay(200)
            println("Second child")
        }
    }
 
    delay(500)
    println("Done")
}

// Channel 接口的简化定义
// Channel 同时实现了发送和接收两个接口
public interface Channel<E> : SendChannel<E>, ReceiveChannel<E>

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 创建一个 Int 类型的通道，默认是 Rendezvous（会合）类型
    val channel = Channel<Int>()
 
    // 启动一个生产者协程
    launch {
        for (x in 1..5) {
            println("发送 $x")  // 打印发送日志
            channel.send(x)     // 挂起式发送，如果通道满了就挂起等待
        }
        channel.close()         // 发送完毕后关闭通道，通知消费者不会再有新数据
    }
 
    // 在主协程中消费数据
    // 使用 for-in 迭代通道，通道关闭后循环自动结束
    for (value in channel) {
        println("接收到 $value")
    }
 
    println("通道已关闭，消费完毕")
}

// SendChannel 接口中的 send 签名
public interface SendChannel<in E> {
    // 向通道发送指定的元素 element
    // 如果通道的缓冲区已满，该函数会挂起，直到有空间可用
    public suspend fun send(element: E)
 
    // 非挂起版本的尝试发送（Kotlin 1.5+ 推荐使用 trySend 替代已废弃的 offer）
    // 立即返回结果，不会挂起
    public fun trySend(element: E): ChannelResult<Unit>
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 创建一个容量为 2 的缓冲通道
    val channel = Channel<String>(capacity = 2)
 
    launch {
        val items = listOf("A", "B", "C", "D")
        for (item in items) {
            println("准备发送 $item ...")
            channel.send(item)  // 缓冲区满时会挂起
            println("已发送 $item")
        }
        channel.close()  // 所有数据发送完毕，关闭通道
    }
 
    // 故意延迟消费，观察 send 的挂起行为
    delay(1000)  // 等待 1 秒，让生产者先跑
    println("--- 开始消费 ---")
    for (value in channel) {
        println("接收到: $value")
        delay(500)  // 模拟消费耗时
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 容量为 1 的缓冲通道
    val channel = Channel<Int>(capacity = 1)
 
    // 第一次 trySend：缓冲区空，发送成功
    val result1 = channel.trySend(1)
    println("trySend(1): ${result1.isSuccess}")  // true
 
    // 第二次 trySend：缓冲区已满，发送失败
    val result2 = channel.trySend(2)
    println("trySend(2): ${result2.isSuccess}")  // false
 
    // 取走一个元素后再试
    channel.receive()  // 取走 1
    val result3 = channel.trySend(2)
    println("trySend(2) 重试: ${result3.isSuccess}")  // true
 
    channel.close()
}

// ReceiveChannel 接口中的核心方法
public interface ReceiveChannel<out E> {
    // 从通道接收一个元素
    // 如果通道为空，该函数会挂起，直到有新元素可用
    // 如果通道已关闭且为空，抛出 ClosedReceiveChannelException
    public suspend fun receive(): E
 
    // 非挂起版本的尝试接收
    public fun tryReceive(): ChannelResult<E>
 
    // 接收并返回一个 ChannelResult，通道关闭时返回 closed 结果而非抛异常
    public suspend fun receiveCatching(): ChannelResult<E>
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    val channel = Channel<Int>()
 
    // 生产者：发送 3 个元素后关闭
    launch {
        for (i in 1..3) {
            channel.send(i)
        }
        channel.close()  // 关闭通道
    }
 
    // === 方式一：for-in 循环（推荐，最简洁） ===
    // 通道关闭后循环自动结束，无需手动判断
    // for (value in channel) {
    //     println(value)
    // }
 
    // === 方式二：receiveCatching（安全接收） ===
    // 返回 ChannelResult，可以区分"成功接收"、"通道关闭"、"失败"
    while (true) {
        val result = channel.receiveCatching()  // 不会抛异常
        if (result.isClosed) {
            println("通道已关闭，退出循环")
            break
        }
        println("接收到: ${result.getOrNull()}")  // 安全获取值
    }
 
    // === 方式三：tryReceive（非挂起，立即返回） ===
    // 适合轮询场景，但通常不推荐，因为会浪费 CPU
    // val result = channel.tryReceive()
    // if (result.isSuccess) { ... }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    val channel = Channel<Int>()
 
    launch {
        channel.send(42)
        channel.close()  // 发送一个元素后立即关闭
    }
 
    // 第一次 receive：成功取到 42
    println(channel.receive())  // 输出: 42
 
    // 第二次 receive：通道已关闭且为空，抛出异常！
    try {
        channel.receive()  // ClosedReceiveChannelException
    } catch (e: ClosedReceiveChannelException) {
        println("捕获异常: ${e.message}")  // Channel was closed
    }
 
    // 安全的做法：使用 receiveCatching
    // val result = channel.receiveCatching()
    // println(result.isClosed)  // true，不会抛异常
}

// SendChannel 接口中的 close 方法
public interface SendChannel<in E> {
    // 关闭通道
    // cause 参数可选，用于传递关闭原因（异常）
    // 关闭后：send 会抛异常，但已在缓冲区中的元素仍可被 receive
    public fun close(cause: Throwable? = null): Boolean
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 容量为 5 的缓冲通道
    val channel = Channel<Int>(capacity = 5)
 
    // 生产者：一口气发送 5 个元素，然后立即关闭
    launch {
        for (i in 1..5) {
            channel.send(i)       // 缓冲区足够大，不会挂起
            println("已发送: $i")
        }
        channel.close()           // 关闭通道
        println("通道已关闭")
 
        // 关闭后再尝试发送
        try {
            channel.send(6)       // 抛出 ClosedSendChannelException
        } catch (e: ClosedSendChannelException) {
            println("关闭后发送失败: ${e.message}")
        }
    }
 
    // 消费者：延迟一会儿再开始消费
    delay(100)
    println("--- 开始消费 ---")
 
    // 即使通道已关闭，缓冲区中的 5 个元素仍然可以被接收
    for (value in channel) {
        println("接收到: $value")
    }
    println("所有数据消费完毕")
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    val channel = Channel<Int>()
 
    launch {
        channel.send(1)
        channel.send(2)
        // 模拟发生错误，带异常关闭通道
        channel.close(IllegalStateException("数据源出错"))
    }
 
    // receiveCatching 可以拿到关闭原因
    while (true) {
        val result = channel.receiveCatching()
        if (result.isClosed) {
            // exceptionOrNull() 可以获取 close 时传入的异常
            val cause = result.exceptionOrNull()
            println("通道关闭，原因: $cause")
            break
        }
        println("接收到: ${result.getOrNull()}")
    }
}

// close()：优雅关闭，已缓冲的数据仍可被消费
// 属于 SendChannel 的方法
// 语义：生产者说"我不会再发了，但你把剩下的消费完"
channel.close()
 
// cancel()：强制取消，丢弃所有未消费的缓冲数据
// 属于 ReceiveChannel 的方法
// 语义：消费者说"我不要了，全部丢掉"
channel.cancel()

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 演示 cancel 会丢弃缓冲数据
    val channel = Channel<Int>(capacity = 10)
 
    launch {
        for (i in 1..10) {
            channel.send(i)  // 全部放入缓冲区
        }
    }
 
    delay(50)  // 等待生产者发送完毕
 
    // 只消费前 3 个
    repeat(3) {
        println("接收到: ${channel.receive()}")
    }
 
    // 消费者不想要剩下的了，直接 cancel
    channel.cancel()  // 缓冲区中剩余的 4~10 全部被丢弃
    println("通道已取消")
 
    // 此时再 receive 会抛 CancellationException
    try {
        channel.receive()
    } catch (e: CancellationException) {
        println("取消后接收失败: ${e.message}")
    }
}

// Channel 内部结构示意（简化）
//
// ┌─────────────────────────────────────────────────┐
// │                  Channel<E>                      │
// │                                                  │
// │  SendChannel<E>          ReceiveChannel<E>       │
// │  ┌──────────────┐       ┌───────────────────┐    │
// │  │ send(e)      │       │ receive(): E      │    │
// │  │ trySend(e)   │       │ tryReceive()      │    │
// │  │ close()      │       │ receiveCatching() │    │
// │  └──────┬───────┘       │ cancel()          │    │
// │         │               └────────┬──────────┘    │
// │         ▼                        ▲               │
// │  ┌──────────────────────────────────────────┐    │
// │  │         Buffer (缓冲区)                   │    │
// │  │  [ E₁ | E₂ | E₃ | ... | Eₙ ]           │    │
// │  │  容量由 Channel(capacity) 决定            │    │
// │  └──────────────────────────────────────────┘    │
// │                                                  │
// │  isClosedForSend: Boolean  // 是否对发送关闭     │
// │  isClosedForReceive: Boolean // 是否对接收关闭   │
// │  (对接收关闭 = 对发送关闭 + 缓冲区为空)          │
// └─────────────────────────────────────────────────┘

fun main() = runBlocking {
    val channel = Channel<Int>(capacity = 2)
    
    launch {
        for (i in 1..4) {
            channel.send(i)
        }
        channel.close()
    }
    
    delay(100)
    val first = channel.receive()
    val second = channel.receive()
    println("$first, $second")
    
    val result = channel.receiveCatching()
    println("${result.getOrNull()}, isClosed=${result.isClosed}")
}

// Channel 伴生对象中的容量常量
val RENDEZVOUS = 0        // 会合通道，容量为 0
val BUFFERED = 64         // 默认缓冲大小（可通过系统属性修改）
val CONFLATED = -1        // 冲突通道，只保留最新值
val UNLIMITED = Int.MAX_VALUE  // 无限容量通道

// 以下两种写法等价，都创建会合通道
val channel1 = Channel<Int>()                    // 默认就是 RENDEZVOUS
val channel2 = Channel<Int>(Channel.RENDEZVOUS)  // 显式指定

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 创建一个会合通道，容量为 0
    val channel = Channel<String>()
 
    // 启动生产者协程
    launch {
        println("[生产者] 准备发送 A...")
        channel.send("A")  // 挂起！因为此时还没有人 receive
        println("[生产者] A 已发送")  // 只有消费者 receive 之后才会打印
 
        println("[生产者] 准备发送 B...")
        channel.send("B")  // 再次挂起，等待下一次 receive
        println("[生产者] B 已发送")
 
        channel.close()  // 发送完毕，关闭通道
    }
 
    // 主协程充当消费者，故意延迟来观察行为
    delay(1000)  // 延迟 1 秒，让生产者先到达 send 并挂起
    println("[消费者] 准备接收第一个值...")
    val first = channel.receive()  // 此时生产者的 send("A") 恢复
    println("[消费者] 收到: $first")
 
    delay(1000)  // 再延迟 1 秒
    println("[消费者] 准备接收第二个值...")
    val second = channel.receive()  // 生产者的 send("B") 恢复
    println("[消费者] 收到: $second")
}

// 会合通道的内存模型（概念图）
//
// 生产者协程                    消费者协程
// ┌──────────┐                ┌──────────┐
// │ send("A")│───── 握手 ─────│ receive()│
// │  挂起... │   直接传递数据  │  挂起... │
// └──────────┘                └──────────┘
//        ↕                          ↕
//   谁先到谁等待              谁先到谁等待
//
// 通道内部缓冲区大小 = 0，没有中间存储

// 创建一个容量为 3 的缓冲通道
val channel = Channel<Int>(capacity = 3)
 
// 使用默认缓冲大小（64）
val defaultBuffered = Channel<Int>(Channel.BUFFERED)

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 创建容量为 3 的缓冲通道
    val channel = Channel<Int>(capacity = 3)
 
    // 生产者：快速发送 5 个元素
    val producer = launch {
        for (i in 1..5) {
            println("[生产者] 发送 $i ...")
            channel.send(i)  // 前 3 个不会挂起，第 4 个开始挂起
            println("[生产者] $i 已进入缓冲区")
        }
        channel.close()
    }
 
    // 消费者：故意慢速消费，每次间隔 1 秒
    val consumer = launch {
        delay(2000)  // 先等 2 秒，让生产者把缓冲区填满
        for (value in channel) {
            println("[消费者] 收到: $value")
            delay(1000)  // 模拟慢速处理
        }
    }
 
    producer.join()
    consumer.join()
}

// 创建无限容量通道
val channel = Channel<Int>(Channel.UNLIMITED)

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 创建无限容量通道
    val channel = Channel<Int>(Channel.UNLIMITED)
 
    // 生产者：瞬间发送大量数据，send 永不挂起
    val producer = launch {
        for (i in 1..1_000_000) {
            channel.send(i)  // 永远不会挂起，全部进入内存队列
        }
        println("[生产者] 100 万条数据全部发送完毕")
        channel.close()
    }
 
    // 消费者：慢速消费
    val consumer = launch {
        var count = 0
        for (value in channel) {
            count++
            if (count % 200_000 == 0) {
                println("[消费者] 已处理 $count 条")
                delay(100)  // 模拟处理耗时
            }
        }
        println("[消费者] 全部处理完毕，共 $count 条")
    }
 
    producer.join()
    consumer.join()
}

// 无限容量通道的内存模型
//
// 生产者                    通道内部队列                    消费者
// ┌──────┐    send()     ┌──────────────────────┐    receive()   ┌──────┐
// │ 快速 │──────────────→│ 1 2 3 4 5 ... 99999  │──────────────→│ 慢速 │
// │ 生产 │   永不挂起     │   无限增长的队列！      │   按需取出     │ 消费 │
// └──────┘               └──────────────────────┘               └──────┘
//                              ↑
//                         内存持续增长
//                         OOM 风险！

// 创建冲突通道
val channel = Channel<Int>(Channel.CONFLATED)

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 创建冲突通道
    val channel = Channel<Int>(Channel.CONFLATED)
 
    // 生产者：快速发送 1 到 10
    launch {
        for (i in 1..10) {
            println("[生产者] 发送 $i")
            channel.send(i)  // 永不挂起，但旧值会被覆盖
        }
        channel.close()
    }
 
    // 消费者：故意延迟，让生产者先把数据全部发完
    delay(500)  // 等待 500ms，生产者早已发完全部数据
 
    // 此时通道中只保留了最后一个值
    for (value in channel) {
        println("[消费者] 收到: $value")
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
// 模拟温度传感器，每 100ms 产生一个读数
fun main() = runBlocking {
    val sensorChannel = Channel<Double>(Channel.CONFLATED)
 
    // 传感器：高频率产生温度数据
    val sensor = launch {
        var temp = 20.0
        repeat(100) {  // 产生 100 个读数
            temp += (-0.5..0.5).random()  // 模拟温度波动
            sensorChannel.send(temp)       // 永不挂起，旧值被覆盖
            delay(100)                     // 每 100ms 一个读数
        }
        sensorChannel.close()
    }
 
    // UI 更新：每 1 秒刷新一次显示（比传感器慢 10 倍）
    val display = launch {
        for (temp in sensorChannel) {
            // 每次拿到的都是最新的温度值
            // 中间被跳过的旧值对 UI 显示没有意义
            println("[显示] 当前温度: ${"%.1f".format(temp)}°C")
            delay(1000)  // UI 每秒刷新一次
        }
    }
 
    sensor.join()
    display.join()
}

// 四种通道类型速查表
//
// ┌──────────────┬────────────┬──────────────┬──────────────┬───────────────┐
// │              │ RENDEZVOUS │   BUFFERED   │  UNLIMITED   │   CONFLATED   │
// │              │  (默认)     │  (固定缓冲)   │  (无限缓冲)   │  (只留最新)    │
// ├──────────────┼────────────┼──────────────┼──────────────┼───────────────┤
// │ capacity     │     0      │   N (默认64)  │ Int.MAX_VALUE│     -1        │
// ├──────────────┼────────────┼──────────────┼──────────────┼───────────────┤
// │ send 挂起?   │ 总是挂起    │ 缓冲满时挂起  │  永不挂起     │   永不挂起     │
// │              │(等receive) │              │              │               │
// ├──────────────┼────────────┼──────────────┼──────────────┼───────────────┤
// │ receive 挂起?│ 总是挂起    │ 缓冲空时挂起  │ 缓冲空时挂起  │  缓冲空时挂起  │
// │              │(等send)    │              │              │               │
// ├──────────────┼────────────┼──────────────┼──────────────┼───────────────┤
// │ 数据丢失?    │    否      │     否       │     否       │    是(旧值)    │
// ├──────────────┼────────────┼──────────────┼──────────────┼───────────────┤
// │ OOM 风险?    │    否      │     否       │     是       │     否        │
// ├──────────────┼────────────┼──────────────┼──────────────┼───────────────┤
// │ 典型场景     │ 严格同步    │ 通用生产消费  │ 事件总线      │  状态/传感器   │
// │              │ 握手传递    │ 速率解耦      │ 有限突发      │  只关心最新值  │
// └──────────────┴────────────┴──────────────┴──────────────┴───────────────┘

import kotlinx.coroutines.channels.*
 
// BufferOverflow 枚举有三个值：
// SUSPEND    — 缓冲区满时 send 挂起（默认行为）
// DROP_OLDEST — 缓冲区满时丢弃最旧的元素，send 不挂起
// DROP_LATEST — 缓冲区满时丢弃最新要发送的元素，send 不挂起
 
// 示例：容量为 3，满了就丢弃最旧的元素
val channel = Channel<Int>(
    capacity = 3,                              // 缓冲区大小为 3
    onBufferOverflow = BufferOverflow.DROP_OLDEST  // 满了丢最旧的
)

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    println("=== DROP_OLDEST 策略 ===")
    testOverflow(BufferOverflow.DROP_OLDEST)
 
    println("\n=== DROP_LATEST 策略 ===")
    testOverflow(BufferOverflow.DROP_LATEST)
}
 
suspend fun testOverflow(policy: BufferOverflow) {
    // 创建容量为 3 的通道，使用指定的溢出策略
    val channel = Channel<Int>(
        capacity = 3,
        onBufferOverflow = policy
    )
 
    // 生产者：快速发送 1~6
    for (i in 1..6) {
        channel.send(i)  // 不会挂起！溢出时按策略丢弃
        println("  发送: $i")
    }
    channel.close()
 
    // 消费者：一次性取出所有剩余元素
    print("  收到: ")
    for (value in channel) {
        print("$value ")
    }
    println()
}

fun main() = runBlocking {
    val channel = Channel<Int>(Channel.CONFLATED)
 
    launch {
        channel.send(1)
        channel.send(2)
        channel.send(3)
        delay(200)
        channel.send(4)
        channel.send(5)
        channel.close()
    }
 
    delay(100)
    println(channel.receive())
    delay(300)
    println(channel.receive())
}

// 手动创建 Channel 并在协程中发送数据
fun CoroutineScope.manualProducer(): Channel<Int> {
    // 手动创建一个 Channel 实例
    val channel = Channel<Int>()
    // 启动一个协程来发送数据
    launch {
        for (i in 1..5) {
            channel.send(i)       // 逐个发送数据
        }
        channel.close()           // ⚠️ 必须手动关闭，否则消费端会永远挂起等待
    }
    return channel                // 返回 Channel 供消费端使用
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // produce 构建器：创建一个生产者协程
    // 返回值类型是 ReceiveChannel<Int>，只暴露接收能力
    val channel: ReceiveChannel<Int> = produce {
        // 这个 lambda 的 receiver 是 ProducerScope<Int>
        // ProducerScope 同时实现了 CoroutineScope 和 SendChannel
        for (i in 1..5) {
            send(i)                // 直接调用 send，无需持有 channel 引用
            println("已发送: $i")  // 打印发送日志
        }
        // ✅ 不需要手动 close()！produce 会在协程结束时自动关闭 Channel
    }
 
    // 消费端：通过 for-in 遍历接收数据
    for (value in channel) {
        println("已接收: $value")
    }
    println("消费完毕")
}

// produce 函数签名
fun <E> CoroutineScope.produce(
    context: CoroutineContext = EmptyCoroutineContext,  // 可选：附加的协程上下文（如 Dispatchers）
    capacity: Int = 0,                                  // Channel 容量，默认 0（Rendezvous）
    block: suspend ProducerScope<E>.() -> Unit          // 生产者逻辑，运行在 ProducerScope 中
): ReceiveChannel<E>                                    // 返回只读的接收通道

fun main() = runBlocking {
    // 指定容量为 3 的缓冲 Channel，运行在 IO 调度器上
    val channel = produce(Dispatchers.IO, capacity = 3) {
        for (i in 1..10) {
            send(i)
            // 因为有缓冲，前 3 个 send 不会挂起
            println("[${Thread.currentThread().name}] 发送: $i")
        }
    }
 
    // 模拟慢消费者
    for (value in channel) {
        delay(200)  // 每 200ms 消费一个
        println("[${Thread.currentThread().name}] 接收: $value")
    }
}

fun main() = runBlocking {
    val channel = produce {
        send(1)
        send(2)
        // 模拟生产过程中发生异常
        throw RuntimeException("生产出错了！")
        // send(3) 永远不会执行
    }
 
    try {
        // 消费端会收到前两个值，然后在下一次 receive 时收到异常
        for (value in channel) {
            println("接收: $value")
        }
    } catch (e: RuntimeException) {
        // Channel 被异常关闭时，消费端会收到这个异常
        println("捕获异常: ${e.message}")
    }
}

fun main() = runBlocking {
    val channel = produce {
        // 在 ProducerScope 内启动多个子协程并行生产
        launch {
            for (i in 1..3) {
                send(i * 10)          // 子协程 1 发送 10, 20, 30
                delay(100)
            }
        }
        launch {
            for (i in 1..3) {
                send(i * 100)         // 子协程 2 发送 100, 200, 300
                delay(150)
            }
        }
        // produce 会等待所有子协程完成后才关闭 Channel
    }
 
    for (value in channel) {
        println("接收: $value")
    }
}

// 方式一：for-in 循环（不会自动取消 Channel）
for (value in channel) {
    // 如果这里抛异常，channel 本身不会被取消
    // 生产者可能还在继续运行、继续发送
    process(value)
}
 
// 方式二：consumeEach（会自动取消 Channel）
channel.consumeEach { value ->
    // 如果这里抛异常，consumeEach 会自动取消（cancel）Channel
    // 生产者会收到 CancellationException，停止生产
    process(value)
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 创建一个生产者
    val numbers = produce {
        for (i in 1..5) {
            send(i)
            println("发送: $i")
        }
    }
 
    // 使用 consumeEach 消费所有数据
    // consumeEach 是一个挂起函数，会在 Channel 关闭后正常返回
    numbers.consumeEach { value ->
        println("消费: $value")
    }
 
    println("所有数据已消费完毕")
}

fun main() = runBlocking {
    // ========== 场景 1：for-in 遇到异常 ==========
    println("=== 场景 1: for-in ===")
    val ch1 = produce {
        for (i in 1..5) {
            send(i)
            println("  [生产者1] 发送: $i")
        }
        // 如果消费端异常但没取消 channel，这里的后续逻辑可能受影响
        println("  [生产者1] 生产完毕")
    }
 
    try {
        for (value in ch1) {
            println("  [消费者1] 接收: $value")
            if (value == 3) throw RuntimeException("消费出错")
        }
    } catch (e: Exception) {
        println("  [消费者1] 捕获异常: ${e.message}")
        // ⚠️ 需要手动取消 channel
        ch1.cancel()
    }
 
    delay(100) // 等待一下观察生产者状态
 
    // ========== 场景 2：consumeEach 遇到异常 ==========
    println("\n=== 场景 2: consumeEach ===")
    val ch2 = produce {
        for (i in 1..5) {
            send(i)
            println("  [生产者2] 发送: $i")
        }
        println("  [生产者2] 生产完毕")
    }
 
    try {
        ch2.consumeEach { value ->
            println("  [消费者2] 接收: $value")
            if (value == 3) throw RuntimeException("消费出错")
            // ✅ consumeEach 内部会自动 cancel channel
        }
    } catch (e: Exception) {
        println("  [消费者2] 捕获异常: ${e.message}")
        // 不需要手动取消，consumeEach 已经处理了
    }
}

// consumeEach 的简化实现（伪代码）
public suspend inline fun <E> ReceiveChannel<E>.consumeEach(action: (E) -> Unit) {
    // consume 是另一个扩展函数，它保证在 block 结束后取消 Channel
    consume {
        // 使用迭代器遍历 Channel 中的所有元素
        for (element in this) {
            action(element)    // 对每个元素执行用户提供的操作
        }
    }
}
 
// consume 的简化实现
public inline fun <E, R> ReceiveChannel<E>.consume(block: ReceiveChannel<E>.() -> R): R {
    var cause: Throwable? = null
    try {
        return block()         // 执行消费逻辑
    } catch (e: Throwable) {
        cause = e              // 记录异常
        throw e                // 重新抛出
    } finally {
        // ✅ 关键：无论正常还是异常，都会取消 Channel
        cancelConsumed(cause)
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
// 模拟一个数据源：从"数据库"中逐条读取用户 ID
fun CoroutineScope.produceUserIds(): ReceiveChannel<Int> = produce(capacity = 5) {
    // capacity = 5 表示缓冲 5 个元素，生产者可以"跑在前面"
    println("[生产者] 开始读取用户 ID...")
    for (id in 1001..1010) {
        delay(50)              // 模拟数据库查询耗时
        send(id)               // 发送用户 ID 到 Channel
        println("[生产者] 发送用户 ID: $id")
    }
    println("[生产者] 所有用户 ID 已发送")
    // ✅ produce 自动关闭 Channel
}
 
// 模拟处理逻辑：根据用户 ID 查询详细信息
suspend fun processUser(id: Int): String {
    delay(100)                 // 模拟网络请求耗时
    return "User($id, name=用户$id)"
}
 
fun main() = runBlocking {
    // 启动生产者
    val userIds = produceUserIds()
 
    // 使用 consumeEach 消费
    userIds.consumeEach { id ->
        val userInfo = processUser(id)   // 处理每个用户 ID
        println("[消费者] 处理完成: $userInfo")
    }
 
    println("全部处理完毕")
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
// 第一阶段：生产原始数据
fun CoroutineScope.produceNumbers(): ReceiveChannel<Int> = produce {
    for (i in 1..10) {
        send(i)                // 发送 1 到 10
    }
}
 
// 第二阶段：过滤 —— 只保留偶数
fun CoroutineScope.filterEven(
    input: ReceiveChannel<Int>       // 接收上一阶段的输出
): ReceiveChannel<Int> = produce {
    // 消费 input 中的每个元素
    input.consumeEach { value ->
        if (value % 2 == 0) {        // 只保留偶数
            send(value)              // 转发到下一阶段
        }
    }
}
 
// 第三阶段：转换 —— 将数字平方
fun CoroutineScope.square(
    input: ReceiveChannel<Int>
): ReceiveChannel<Int> = produce {
    input.consumeEach { value ->
        send(value * value)          // 平方后转发
    }
}
 
fun main() = runBlocking {
    // 构建处理链：生产 -> 过滤偶数 -> 平方
    val numbers = produceNumbers()          // 1, 2, 3, ..., 10
    val evenNumbers = filterEven(numbers)   // 2, 4, 6, 8, 10
    val squaredNumbers = square(evenNumbers) // 4, 16, 36, 64, 100
 
    // 最终消费
    squaredNumbers.consumeEach { result ->
        println("结果: $result")
    }
}

fun main() = runBlocking {
    val channel = produce {
        var count = 0
        // 持续生产数据，直到被取消
        while (true) {
            send(count++)
            delay(200)             // 每 200ms 生产一个
        }
    }
 
    // 只消费 1 秒钟的数据
    withTimeout(1000) {
        channel.consumeEach { value ->
            println("接收: $value")
        }
    }
    // withTimeout 超时后会取消内部协程
    // consumeEach 会取消 channel
    // 生产者协程收到 CancellationException 后停止
    println("超时，停止消费")
}

// ❌ 错误：在 produce 外部持有 SendChannel 引用并发送
val channel = produce<Int> { /* ... */ }
// channel 的类型是 ReceiveChannel<Int>，无法 send
// 这是设计上的刻意限制，保证生产逻辑封装在 produce 内部
 
// ❌ 错误：对同一个 ReceiveChannel 多次调用 consumeEach
val ch = produce { send(1); send(2); send(3) }
ch.consumeEach { println(it) }   // 第一次：正常消费
ch.consumeEach { println(it) }   // 第二次：Channel 已关闭/取消，不会收到任何数据
 
// ✅ 正确：一个 produce 对应一个消费者
// 如果需要多个消费者，应该使用 BroadcastChannel 或 SharedFlow
 
// ❌ 注意：consumeEach 内部不要再调用 channel.cancel()
// consumeEach 已经在 finally 中处理了取消逻辑，重复取消虽然不会报错但没有意义

fun main() = runBlocking {
    val job = launch {
        val channel = produce {
            var i = 0
            while (true) {
                send(i++)
                delay(100)
            }
        }
 
        channel.consumeEach {
            println("接收: $it")
        }
    }
 
    delay(500)
    job.cancel()       // 取消父协程 → 生产者协程也被取消 → Channel 关闭
    println("已取消")
}

fun main() = runBlocking {
    val channel = produce {
        send(1)
        send(2)
        send(3)
    }
 
    println(channel.receive())
    channel.consumeEach { println(it) }
    println("done")
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
// 阶段一：生产者，生成 1 到 5 的整数
fun CoroutineScope.produceNumbers(): ReceiveChannel<Int> = produce {
    for (i in 1..5) {       // 循环生成 1, 2, 3, 4, 5
        send(i)             // 将每个数字发送到通道
        println("生产: $i") // 打印日志，观察执行顺序
    }
    // produce 构建器在 lambda 结束后自动关闭通道
}
 
// 阶段二：变换器，接收上游通道的数据，计算平方后发送到下游
fun CoroutineScope.square(input: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (value in input) {          // 从上游通道逐个接收数据
        val result = value * value  // 计算平方
        send(result)                // 将结果发送到下游通道
        println("变换: $value -> $result")
    }
    // 上游通道关闭后，for 循环自动结束，本通道也随之关闭
}
 
fun main() = runBlocking {
    // 构建管道：将两个阶段串联起来
    val numbers = produceNumbers()      // 阶段一的输出通道
    val squared = square(numbers)       // 阶段二以阶段一的输出作为输入
 
    // 阶段三：最终消费者，直接在主协程中消费
    for (result in squared) {           // 从最终通道逐个接收结果
        println("消费: $result")
    }
 
    println("管道处理完毕")
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
// 模拟的原始日志数据类
data class RawLog(
    val timestamp: Long,    // 时间戳
    val level: String,      // 日志级别: INFO, WARN, ERROR
    val message: String     // 日志内容
)
 
// 解析后的结构化日志
data class ParsedLog(
    val time: String,       // 格式化后的时间
    val level: String,      // 日志级别
    val message: String,    // 日志内容
    val isError: Boolean    // 是否为错误日志
)
 
// 阶段一：日志生产者，模拟不断产生的原始日志
fun CoroutineScope.produceRawLogs(): ReceiveChannel<RawLog> = produce {
    // 模拟一批原始日志数据
    val logs = listOf(
        RawLog(1000L, "INFO", "Application started"),
        RawLog(2000L, "DEBUG", "Loading config"),       // DEBUG 级别，后续会被过滤
        RawLog(3000L, "WARN", "Memory usage high"),
        RawLog(4000L, "ERROR", "Database connection failed"),
        RawLog(5000L, "INFO", "Retrying connection"),
        RawLog(6000L, "ERROR", "Retry failed"),
        RawLog(7000L, "DEBUG", "Dumping stack trace"),  // DEBUG 级别，后续会被过滤
        RawLog(8000L, "INFO", "Application shutdown")
    )
    for (log in logs) {     // 逐条发送原始日志
        send(log)
    }
}
 
// 阶段二：过滤器，只保留 INFO、WARN、ERROR 级别的日志（过滤掉 DEBUG）
fun CoroutineScope.filterLogs(
    input: ReceiveChannel<RawLog>   // 上游通道：原始日志
): ReceiveChannel<RawLog> = produce {
    for (log in input) {                        // 从上游逐条接收
        if (log.level != "DEBUG") {             // 过滤条件：排除 DEBUG
            send(log)                           // 符合条件的日志发送到下游
        }
        // 不符合条件的日志被静默丢弃
    }
}
 
// 阶段三：解析器，将原始日志转换为结构化的 ParsedLog
fun CoroutineScope.parseLogs(
    input: ReceiveChannel<RawLog>   // 上游通道：过滤后的原始日志
): ReceiveChannel<ParsedLog> = produce {
    for (log in input) {                        // 从上游逐条接收
        // 将时间戳转换为可读的时间字符串（简化处理）
        val timeStr = "T+${log.timestamp / 1000}s"
        // 构建结构化日志对象
        val parsed = ParsedLog(
            time = timeStr,                     // 格式化时间
            level = log.level,                  // 保留原始级别
            message = log.message,              // 保留原始消息
            isError = log.level == "ERROR"      // 标记是否为错误
        )
        send(parsed)                            // 发送到下游
    }
}
 
// 阶段四：格式化输出器，将结构化日志转换为最终的展示字符串
fun CoroutineScope.formatLogs(
    input: ReceiveChannel<ParsedLog>    // 上游通道：结构化日志
): ReceiveChannel<String> = produce {
    for (log in input) {                        // 从上游逐条接收
        // 根据是否为错误日志，添加不同的前缀标记
        val prefix = if (log.isError) "🔴" else "🟢"
        // 拼接最终的格式化字符串
        val formatted = "$prefix [${log.time}] ${log.level}: ${log.message}"
        send(formatted)                         // 发送到下游
    }
}
 
fun main() = runBlocking {
    // 构建四阶段管道，每个阶段的输出作为下一个阶段的输入
    val raw = produceRawLogs()          // 阶段一 → 输出原始日志
    val filtered = filterLogs(raw)      // 阶段二 → 输出过滤后的日志
    val parsed = parseLogs(filtered)    // 阶段三 → 输出结构化日志
    val formatted = formatLogs(parsed)  // 阶段四 → 输出格式化字符串
 
    // 最终消费：打印所有格式化后的日志
    for (line in formatted) {
        println(line)
    }
}

// 这段代码展示了关闭传播的机制
fun CoroutineScope.stage(input: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (value in input) {  // 当 input 被关闭且所有元素被消费完毕后
        send(value * 2)     // 这个循环自然结束
    }
    // produce 块结束 → 输出通道自动关闭 → 下游感知到关闭
    println("阶段结束，通道已关闭")
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
// 快速生产者
fun CoroutineScope.fastProducer(): ReceiveChannel<Int> = produce {
    for (i in 1..10) {
        println("即将发送: $i")
        send(i)                         // 如果下游还没消费，这里会挂起等待
        println("已发送: $i")
    }
}
 
// 慢速消费的变换阶段
fun CoroutineScope.slowTransform(
    input: ReceiveChannel<Int>
): ReceiveChannel<String> = produce {
    for (value in input) {
        delay(500)                      // 模拟耗时处理（500ms）
        send("处理结果: $value")        // 处理完毕后发送
    }
}
 
fun main() = runBlocking {
    val fast = fastProducer()           // 生产者很快
    val slow = slowTransform(fast)      // 变换阶段很慢
 
    for (result in slow) {              // 最终消费
        println(result)
    }
    // 观察输出会发现：生产者不会一口气发完 10 个
    // 而是发一个、等下游消费一个、再发下一个
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
// 使用 produce 的 capacity 参数来指定输出通道的缓冲大小
fun CoroutineScope.bufferedProducer(): ReceiveChannel<Int> = produce(capacity = 5) {
    // 输出通道容量为 5，生产者可以"预生产"最多 5 个元素
    for (i in 1..20) {
        send(i)                 // 通道未满时不会挂起，可以持续生产
        println("生产: $i")
    }
}
 
fun CoroutineScope.bufferedTransform(
    input: ReceiveChannel<Int>
): ReceiveChannel<Int> = produce(capacity = 3) {
    // 变换阶段也有 3 个元素的缓冲
    for (value in input) {
        delay(100)              // 模拟处理耗时
        send(value * value)     // 处理后发送到缓冲通道
    }
}
 
fun main() = runBlocking {
    val produced = bufferedProducer()
    val transformed = bufferedTransform(produced)
 
    for (result in transformed) {
        delay(200)              // 消费者比较慢
        println("消费: $result")
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun CoroutineScope.riskyTransform(
    input: ReceiveChannel<Int>
): ReceiveChannel<Int> = produce {
    for (value in input) {
        if (value == 3) {
            // 模拟处理第 3 个元素时出错
            throw IllegalStateException("处理元素 $value 时发生错误")
        }
        send(value * 10)
    }
}
 
fun main() = runBlocking {
    val numbers = produce {
        for (i in 1..5) send(i)     // 生产 1 到 5
    }
 
    val transformed = riskyTransform(numbers)
 
    try {
        for (result in transformed) {   // 消费变换后的数据
            println("收到: $result")
        }
    } catch (e: IllegalStateException) {
        // 当 riskyTransform 抛出异常时
        // 下游的 for 循环会收到这个异常
        println("管道异常: ${e.message}")
    }
 
    // 注意：上游的 numbers 通道中可能还有未消费的数据
    // produce 构建器会在其协程被取消时自动关闭通道
    // 但如果需要确保资源清理，可以显式 cancel
    numbers.cancel()    // 显式取消上游，确保资源释放
}