Channel ⭐

// ========== 冷流 Flow：collect 时才执行 ==========
val coldFlow = flow {
    println("Flow: 开始生产")       // 只有 collect 时才会打印
    emit(1)
    emit(2)
}
// 此时什么都不会发生，因为没有人 collect
 
coldFlow.collect { value ->          // 此刻才触发 "Flow: 开始生产"
    println("收到: $value")
}
 
// ========== 热流 Channel：创建后即可发送 ==========
val channel = Channel<Int>()         // Channel 一创建就"活着"了
 
launch {
    println("Channel: 开始生产")     // 这个协程会立刻启动
    channel.send(1)                  // 发送数据，不管有没有人接收
    channel.send(2)
    channel.close()                  // 手动关闭通道
}
 
// 即使消费者稍后才启动，数据也不会丢失（会挂起等待）
launch {
    for (value in channel) {         // 用 for-in 迭代接收
        println("收到: $value")
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
// 模拟图片 URL
data class ImageUrl(val url: String)
// 模拟下载后的图片数据
data class ImageData(val url: String, val bytes: ByteArray)
 
fun main() = runBlocking {
 
    // 创建一个容量为 5 的 Channel，作为"待下载队列"
    val downloadQueue = Channel<ImageUrl>(capacity = 5)
 
    // ========== 生产者：不断向队列投递下载任务 ==========
    launch {
        val urls = listOf(
            "https://example.com/img1.png",  // 模拟 URL 列表
            "https://example.com/img2.png",
            "https://example.com/img3.png",
            "https://example.com/img4.png",
            "https://example.com/img5.png",
        )
        for (url in urls) {
            println("📤 投递下载任务: $url")
            downloadQueue.send(ImageUrl(url))  // 如果缓冲区满了，生产者会挂起等待
        }
        downloadQueue.close()                  // 所有任务投递完毕，关闭 Channel
        println("📤 生产者：所有任务已投递，Channel 已关闭")
    }
 
    // ========== 消费者 1：下载协程 ==========
    launch {
        for (task in downloadQueue) {          // Channel 关闭且无数据时，循环自动结束
            println("⬇️  [Worker-1] 正在下载: ${task.url}")
            delay(300)                          // 模拟网络耗时
            println("✅  [Worker-1] 下载完成: ${task.url}")
        }
    }
 
    // ========== 消费者 2：另一个下载协程（Fan-out 并行消费）==========
    launch {
        for (task in downloadQueue) {          // 同一个 Channel，竞争获取任务
            println("⬇️  [Worker-2] 正在下载: ${task.url}")
            delay(500)                          // 模拟网络耗时（比 Worker-1 慢）
            println("✅  [Worker-2] 下载完成: ${task.url}")
        }
    }
}

// ======= Java BlockingQueue：阻塞线程 =======
// 线程被阻塞时，它什么也做不了，CPU 资源被白白占用
BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(2);
queue.put(1);   // 如果队列满了 → 当前线程被阻塞（Thread.sleep 级别）
queue.take();   // 如果队列空了 → 当前线程被阻塞

// ======= Kotlin Channel：挂起协程 =======
// 协程被挂起时，底层线程被释放，可以去执行其他协程
val channel = Channel<Int>(capacity = 2)
channel.send(1)    // 如果缓冲区满了 → 当前协程挂起（线程被释放）
channel.receive()  // 如果缓冲区空了 → 当前协程挂起（线程被释放）

// 用 ASCII 图理解"挂起 vs 阻塞"的线程占用差异

┌─────────────────── BlockingQueue（线程模型）───────────────────┐
│                                                                │
│  Thread-1: [put()...阻塞等待💤] ──────────────────────────>    │
│  Thread-2: [put()...阻塞等待💤] ──────────────────────────>    │
│  Thread-3: [take()..阻塞等待💤] ──────────────────────────>    │
│                                                                │
│  ❌ 3 个线程全部被占用，即使什么有效工作都没做                    │
└────────────────────────────────────────────────────────────────┘
 
┌─────────────────── Channel（协程模型）────────────────────────┐
│                                                                │
│  Thread-1: [coroutine-A send()挂起] → [切换到 coroutine-D] →  │
│  Thread-2: [coroutine-B send()挂起] → [切换到 coroutine-E] →  │
│                                                                │
│  ✅ 2 个线程，但同时服务了 A/B/D/E 等多个协程                   │
│  挂起的协程不占用线程，只保留一小块 Continuation 状态             │
└────────────────────────────────────────────────────────────────┘

val channel = Channel<Int>(capacity = 3)
 
launch {
    channel.send(1)                    // 正常发送
    channel.send(2)                    // 正常发送
    channel.close()                    // 关闭 Channel：通知消费者不会再有新数据了
    // channel.send(3)                 // ❌ 关闭后再 send 会抛出 ClosedSendChannelException
}
 
launch {
    for (value in channel) {           // close() 后，迭代会在取完剩余数据后自动结束
        println("收到: $value")         // 打印 1, 2 后，循环正常退出
    }
    println("Channel 已关闭，消费结束")  // 这行一定会执行到
}

// Channel 伴生对象中的常量定义
public companion object Factory {
    const val RENDEZVOUS = 0        // 容量为 0，无缓冲
    const val BUFFERED = 64         // 默认缓冲大小（可通过系统属性覆盖）
    const val UNLIMITED = Int.MAX_VALUE // 无限缓冲
    const val CONFLATED = -1        // 合并，只保留最新值
}

  ┌──────────────────── RENDEZVOUS Channel（capacity = 0） ────────────────────┐
  │                                                                            │
  │   Producer                    Channel(空管道)             Consumer          │
  │   ┌──────┐                    ┌──────────┐               ┌──────┐          │
  │   │send()│──── 挂起等待 ─────▶│ 无缓冲区 │◀── 挂起等待 ──│recv()│          │
  │   └──────┘                    └──────────┘               └──────┘          │
  │                                    │                                       │
  │              双方同时就绪时 ──▶ 直接交接（hand-off）                         │
  │                                                                            │
  └────────────────────────────────────────────────────────────────────────────┘

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
 
    // 创建一个 RENDEZVOUS Channel，capacity 默认就是 0
    val channel = Channel<Int>() // 等价于 Channel<Int>(Channel.RENDEZVOUS)
 
    // 启动生产者协程
    launch {
        for (i in 1..5) {
            println("📤 准备发送: $i  (时间: ${System.currentTimeMillis() % 10000}ms)")
            channel.send(i) // 若无人接收，此处会挂起
            println("✅ 已发送: $i    (时间: ${System.currentTimeMillis() % 10000}ms)")
        }
        channel.close() // 发送完毕后关闭 Channel
    }
 
    // 启动消费者协程
    launch {
        for (value in channel) { // 迭代接收，Channel 关闭后循环自动结束
            println("📥 收到: $value       (时间: ${System.currentTimeMillis() % 10000}ms)")
            delay(1000) // 模拟消费者处理耗时 1 秒
        }
    }
}

// 通过 JVM 参数修改全局默认缓冲大小
// -Dkotlinx.coroutines.channels.defaultBuffer=128
 
// 在代码中读取当前默认值（仅供了解，非公开 API）
// 内部实现: val CHANNEL_DEFAULT_CAPACITY = systemProp(
//     "kotlinx.coroutines.channels.defaultBuffer",
//     64,       // 默认值
//     1,        // 最小值
//     Int.MAX_VALUE - 1 // 最大值
// )

// 方式 1：使用预定义常量（容量 = 64，或被系统属性覆盖后的值）
val ch1 = Channel<Int>(Channel.BUFFERED)
 
// 方式 2：显式指定容量（不受系统属性影响）
val ch2 = Channel<Int>(capacity = 10) // 固定缓冲大小为 10

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
 
    // 创建一个缓冲大小为 3 的 Channel（用小缓冲更容易观察效果）
    val channel = Channel<Int>(capacity = 3)
 
    // 生产者：快速发送 6 个元素
    launch {
        for (i in 1..6) {
            println("📤 准备发送: $i")
            channel.send(i) // 前 3 个不会挂起；第 4 个开始可能挂起
            println("✅ 已发送: $i")
        }
        channel.close()
    }
 
    // 消费者：慢速接收（每次间隔 1 秒）
    launch {
        delay(2000) // 故意延迟 2 秒再开始接收
        for (value in channel) {
            println("📥 收到: $value")
            delay(1000)
        }
    }
}

📤 准备发送: 1
✅ 已发送: 1        ← 缓冲区: [1]，未满，立即返回
📤 准备发送: 2
✅ 已发送: 2        ← 缓冲区: [1, 2]，未满
📤 准备发送: 3
✅ 已发送: 3        ← 缓冲区: [1, 2, 3]，刚好满
📤 准备发送: 4      ← 缓冲区已满！send(4) 挂起......
                     ← （等待 2 秒后消费者启动，取走元素 1）
✅ 已发送: 4        ← 腾出空间，send(4) 恢复
📥 收到: 1
...（后续交替进行）

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
 
    // 创建一个无限缓冲的 Channel
    val channel = Channel<Int>(Channel.UNLIMITED)
 
    // 生产者：瞬间发送大量数据
    launch {
        for (i in 1..1_000_000) {
            channel.send(i) // 永远不会挂起，全部堆积在内存中
        }
        println("📤 已发送 1,000,000 个元素")
        channel.close()
    }
 
    // 消费者：慢速消费
    launch {
        var count = 0
        for (value in channel) {
            count++
            if (count % 200_000 == 0) {
                println("📥 已消费: $count 个元素")
            }
        }
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
 
    // 创建一个 CONFLATED Channel
    val channel = Channel<String>(Channel.CONFLATED)
 
    // 生产者：连续快速发送传感器读数
    launch {
        val readings = listOf("温度: 20°C", "温度: 21°C", "温度: 22°C", "温度: 23°C", "温度: 24°C")
        for (reading in readings) {
            println("📤 发送: $reading")
            channel.send(reading) // 永远不挂起，新值覆盖旧值
        }
        channel.close()
    }
 
    // 消费者：故意延迟后接收
    launch {
        delay(100) // 等生产者全部发完
        for (value in channel) {
            println("📥 收到: $value") // 大概率只收到 "温度: 24°C"
        }
    }
}

📤 发送: 温度: 20°C
📤 发送: 温度: 21°C
📤 发送: 温度: 22°C
📤 发送: 温度: 23°C
📤 发送: 温度: 24°C
📥 收到: 温度: 24°C      ← 只有最新的值被消费者看到！

import kotlinx.coroutines.channels.*
 
// 快速创建各类型 Channel 的惯用写法
fun main() {
    // 1. RENDEZVOUS：严格同步
    val rendezvous = Channel<Int>()                    // 默认即 RENDEZVOUS
 
    // 2. BUFFERED：平衡方案
    val buffered    = Channel<Int>(Channel.BUFFERED)   // 系统默认 64
    val buffered10  = Channel<Int>(10)                 // 自定义缓冲大小
 
    // 3. UNLIMITED：无限缓冲
    val unlimited   = Channel<Int>(Channel.UNLIMITED)  // 谨慎使用
 
    // 4. CONFLATED：只保留最新
    val conflated   = Channel<Int>(Channel.CONFLATED)  // 丢弃旧值
}

fun main() = runBlocking {
    val ch = Channel<Int>(Channel.CONFLATED)
    launch {
        ch.send(1)
        ch.send(2)
        ch.send(3)
        ch.close()
    }
    delay(500)
    for (v in ch) { print("$v ") }
}

// SendChannel 接口中的核心方法
public suspend fun send(element: E)

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 创建容量为 2 的缓冲 Channel
    val channel = Channel<Int>(capacity = 2)
 
    // 启动生产者协程
    val producer = launch {
        for (i in 1..5) {
            println("📤 准备发送: $i")
            // 当缓冲区已满时(已有2个元素)，send 会挂起
            channel.send(i)
            // 只有 send 成功返回后，才会执行这一行
            println("✅ 已发送: $i")
        }
        // 发送完毕后关闭 Channel，通知消费者不再有新数据
        channel.close()
    }
 
    // 模拟慢速消费者：每 500ms 消费一个
    launch {
        // 使用 for-in 迭代，Channel 关闭后自动退出循环
        for (value in channel) {
            println("📥 接收到: $value")
            // 故意延迟，让生产者体验到 send 挂起
            delay(500)
        }
        println("🏁 Channel 已关闭，消费结束")
    }
}

// trySend 不是挂起函数，它返回一个 ChannelResult
// 表明发送操作的成功或失败
public fun trySend(element: E): ChannelResult<Unit>

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 创建一个容量仅为 1 的 Channel
    val channel = Channel<String>(capacity = 1)
 
    // 第一次 trySend：缓冲区为空，发送成功
    val result1 = channel.trySend("Hello")
    // isSuccess = true，元素已放入缓冲区
    println("第一次: isSuccess=${result1.isSuccess}")
 
    // 第二次 trySend：缓冲区已满，发送失败（不会挂起）
    val result2 = channel.trySend("World")
    // isSuccess = false，缓冲区没有空间
    println("第二次: isSuccess=${result2.isSuccess}")
    // isFailure = true，可以检查失败原因
    println("第二次: isFailure=${result2.isFailure}")
 
    // 取出一个元素后再试
    channel.receive()
    val result3 = channel.trySend("World")
    // 此时缓冲区重新有空间，发送成功
    println("第三次: isSuccess=${result3.isSuccess}")
 
    channel.close()
 
    // Channel 关闭后尝试发送
    val result4 = channel.trySend("!")
    // isClosed = true，表明 Channel 已关闭
    println("第四次: isClosed=${result4.isClosed}")
}

// ReceiveChannel 接口中的核心方法
public suspend fun receive(): E

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // 无缓冲 Channel（RENDEZVOUS），send 和 receive 必须"握手"
    val channel = Channel<Int>()
 
    // 消费者先启动，此时 Channel 为空，receive 会挂起
    val consumer = launch {
        println("👂 消费者: 等待第一个元素...")
        // 此处挂起，直到生产者调用 send
        val first = channel.receive()
        println("📥 消费者: 收到 $first")
 
        println("👂 消费者: 等待第二个元素...")
        val second = channel.receive()
        println("📥 消费者: 收到 $second")
    }
 
    // 延迟 1 秒后，生产者才开始发送
    launch {
        delay(1000)
        println("📤 生产者: 发送 42")
        // 消费者已在等待，send 和 receive 同时完成（rendezvous）
        channel.send(42)
 
        delay(1000)
        println("📤 生产者: 发送 99")
        channel.send(99)
    }
}

// tryReceive 不会挂起，立即返回 ChannelResult
public fun tryReceive(): ChannelResult<E>

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    val channel = Channel<Int>(capacity = 3)
 
    // 先放入两个元素
    channel.send(10)
    channel.send(20)
 
    // tryReceive 立即成功，取出 10
    val r1 = channel.tryReceive()
    // getOrNull() 在成功时返回元素，失败时返回 null
    println("第一次: ${r1.getOrNull()}")   // 输出: 10
 
    // tryReceive 立即成功，取出 20
    val r2 = channel.tryReceive()
    println("第二次: ${r2.getOrNull()}")   // 输出: 20
 
    // 缓冲区为空，tryReceive 立即返回失败（不挂起）
    val r3 = channel.tryReceive()
    println("第三次: ${r3.getOrNull()}")   // 输出: null
    println("第三次 isSuccess: ${r3.isSuccess}")  // 输出: false
 
    channel.close()
 
    // Channel 关闭后 tryReceive
    val r4 = channel.tryReceive()
    // isClosed = true，且缓冲区已空
    println("第四次 isClosed: ${r4.isClosed}")  // 输出: true
}

// 返回 ChannelResult<E> 而非直接返回 E
// 不会因 Channel 关闭而抛异常
public suspend fun receiveCatching(): ChannelResult<E>

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    val channel = Channel<String>(capacity = 2)
 
    // 放入两个元素后立刻关闭
    channel.send("Alpha")
    channel.send("Beta")
    channel.close()
 
    // 使用 while(true) + receiveCatching 循环消费
    while (true) {
        // receiveCatching 在 Channel 关闭且缓冲区为空时，
        // 不会抛异常，而是返回一个 closed 的 ChannelResult
        val result = channel.receiveCatching()
 
        // 检查是否已关闭且无数据
        if (result.isClosed) {
            println("🏁 Channel 关闭，退出循环")
            // 如果关闭时有异常原因，可以通过 exceptionOrNull() 获取
            println("关闭原因: ${result.exceptionOrNull()}")
            break
        }
 
        // 成功取出元素
        println("📥 收到: ${result.getOrNull()}")
    }
    // 输出:
    // 📥 收到: Alpha
    // 📥 收到: Beta
    // 🏁 Channel 关闭，退出循环
    // 关闭原因: null
}

// 推荐写法：for-in 迭代
for (element in channel) {
    // 处理元素
    // Channel 关闭后自动退出循环，无需手动检查
    process(element)
}

// close 不是挂起函数，可以在任何地方调用
public fun close(cause: Throwable? = null): Boolean

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    val channel = Channel<Int>(capacity = 3)
 
    // 发送 3 个元素填满缓冲区
    channel.send(1)
    channel.send(2)
    channel.send(3)
 
    // 关闭 Channel，返回 true 表示是首次关闭
    val firstClose = channel.close()
    println("首次 close 返回: $firstClose")    // true
 
    // 重复关闭返回 false，不会抛异常
    val secondClose = channel.close()
    println("重复 close 返回: $secondClose")   // false
 
    // ====== 关闭后发送 ======
    try {
        // Channel 已关闭，send 会立即抛出 ClosedSendChannelException
        channel.send(4)
    } catch (e: ClosedSendChannelException) {
        println("❌ 发送失败: ${e::class.simpleName}")
    }
 
    // ====== 关闭后接收（缓冲区仍有数据）======
    // 缓冲区中的 1, 2, 3 仍然可以正常取出
    println("📥 ${channel.receive()}")  // 1
    println("📥 ${channel.receive()}")  // 2
    println("📥 ${channel.receive()}")  // 3
 
    // ====== 缓冲区清空后再接收 ======
    try {
        // 缓冲区已空且 Channel 已关闭 → 抛出异常
        channel.receive()
    } catch (e: ClosedReceiveChannelException) {
        println("❌ 接收失败: ${e::class.simpleName}")
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    val channel = Channel<Int>(capacity = 2)
 
    channel.send(100)
 
    // 用自定义异常关闭 Channel，表示"出错了"
    channel.close(IllegalStateException("数据源异常，被迫关闭"))
 
    // 缓冲区中的元素仍然可以正常取出
    println("📥 ${channel.receive()}")  // 100
 
    // 缓冲区已空，再次接收时异常会携带 close 的 cause
    val result = channel.receiveCatching()
    println("isClosed: ${result.isClosed}")          // true
    // exceptionOrNull() 返回的就是 close 时传入的异常
    println("原因: ${result.exceptionOrNull()}")
    // 输出: 原因: java.lang.IllegalStateException: 数据源异常，被迫关闭
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    val channel = Channel<Int>(capacity = 2)
 
    channel.send(1)
    channel.send(2)
 
    // 此时 Channel 处于 OPEN 状态
    println("关闭前:")
    // isClosedForSend: 对发送者来说是否已关闭
    println("  isClosedForSend: ${channel.isClosedForSend}")       // false
    // isClosedForReceive: 对接收者来说是否已关闭
    println("  isClosedForReceive: ${channel.isClosedForReceive}") // false
 
    // 关闭 Channel
    channel.close()
 
    println("关闭后(缓冲区有数据):")
    // 对发送者来说，close() 后立即为 true
    println("  isClosedForSend: ${channel.isClosedForSend}")       // true
    // 对接收者来说，缓冲区仍有数据，所以还是 false！
    println("  isClosedForReceive: ${channel.isClosedForReceive}") // false
 
    // 取出所有剩余元素
    channel.receive() // 取出 1
    channel.receive() // 取出 2
 
    println("缓冲区清空后:")
    println("  isClosedForSend: ${channel.isClosedForSend}")       // true
    // 缓冲区已空，对接收者也关闭了
    println("  isClosedForReceive: ${channel.isClosedForReceive}") // true
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    val channel = Channel<Int>(capacity = 5)
 
    // 放入 5 个元素
    repeat(5) { channel.send(it) }
 
    // cancel 会立即丢弃缓冲区中所有未消费的元素
    channel.cancel()
 
    // 不同于 close()：cancel 后即使缓冲区有数据也无法接收
    val result = channel.receiveCatching()
    println("isClosed: ${result.isClosed}")  // true
    // 缓冲区的 0,1,2,3,4 全部丢失
}

close()  → "我说完了，你把剩下的听完。"（优雅关闭，保留缓冲数据）
cancel() → "我不要了，全部扔掉。"    （强制取消，丢弃缓冲数据）

fun main() = runBlocking {
    val ch = Channel<Int>(capacity = 2)
    ch.send(10)
    ch.send(20)
    ch.close()
    
    println(ch.receiveCatching().getOrNull())
    println(ch.receiveCatching().getOrNull())
    println(ch.receiveCatching().getOrNull())
    println(ch.isClosedForReceive)
}

// CoroutineScope 的扩展函数，返回 ReceiveChannel
public fun <E> CoroutineScope.produce(
    context: CoroutineContext = EmptyCoroutineContext,  // 可指定调度器等上下文
    capacity: Int = 0,                                  // Channel 缓冲容量，默认 RENDEZVOUS
    block: suspend ProducerScope<E>.() -> Unit           // 生产者逻辑，lambda 的 receiver 是 ProducerScope
): ReceiveChannel<E>

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun main() = runBlocking {
    // produce 启动一个新协程，并返回 ReceiveChannel<Int>
    val numbers: ReceiveChannel<Int> = produce {
        // 这里的 this 是 ProducerScope<Int>
        for (i in 1..5) {
            println("正在发送: $i")  // 打印发送日志
            send(i)                  // 向 Channel 发送数据
        }
        // lambda 执行完毕后，Channel 自动关闭
        println("生产者协程结束，Channel 已自动关闭")
    }
 
    // 消费者侧：通过 for-in 遍历 ReceiveChannel
    for (num in numbers) {            // 当 Channel 关闭后，for 循环自动退出
        println("接收到: $num")
    }
 
    println("消费完毕")
}

正在发送: 1
接收到: 1
正在发送: 2
接收到: 2
正在发送: 3
接收到: 3
正在发送: 4
接收到: 4
正在发送: 5
接收到: 5
生产者协程结束，Channel 已自动关闭
消费完毕

fun main() = runBlocking {
    val channel = Channel<Int>()      // 手动创建 Channel
 
    // 手动启动生产者协程
    launch {
        for (i in 1..5) {
            channel.send(i)           // 发送数据
        }
        channel.close()               // ⚠️ 必须手动关闭！忘记就会导致消费者永远挂起
    }
 
    for (num in channel) {
        println("接收到: $num")
    }
}

fun main() = runBlocking {
    val channel = Channel<Int>()
 
    launch {
        for (i in 1..5) {
            if (i == 3) throw RuntimeException("出错了！") // 第3次发送时崩溃
            channel.send(i)
        }
        channel.close() // ❌ 这行永远不会被执行！
    }
 
    // 消费者将在接收完 1, 2 后永远挂起
    for (num in channel) {
        println("接收到: $num")
    }
}

launch {
    try {
        for (i in 1..5) {
            if (i == 3) throw RuntimeException("出错了！")
            channel.send(i)
        }
    } finally {
        channel.close() // 无论如何都关闭
    }
}

fun main() = runBlocking {
    val numbers = produce {
        for (i in 1..5) {
            if (i == 3) throw RuntimeException("出错了！")
            send(i)
        }
        // 不需要 try-finally，produce 会自动用异常关闭 Channel
    }
 
    try {
        for (num in numbers) {         // 当异常传播时，这里会抛出异常
            println("接收到: $num")
        }
    } catch (e: Exception) {
        println("捕获异常: ${e.message}")
    }
}

接收到: 1
接收到: 2
捕获异常: 出错了！

fun main() = runBlocking {
    // 创建一个带 5 个缓冲位的 Channel
    val bufferedChannel = produce(capacity = 5) {
        for (i in 1..10) {
            println("[${System.currentTimeMillis()}] 发送: $i")
            send(i)   // 前 5 个不会挂起，因为缓冲区有空间
        }
    }
 
    delay(1000)  // 消费者延迟 1 秒后才开始消费
 
    for (num in bufferedChannel) {
        println("[${System.currentTimeMillis()}] 接收: $num")
    }
}

// RENDEZVOUS (默认, capacity = 0):
//   每次 send 必须等 receive 到来才能继续
produce(capacity = Channel.RENDEZVOUS) { /* ... */ }
 
// BUFFERED (capacity = Channel.BUFFERED, 默认64):
//   缓冲区满之前 send 不挂起
produce(capacity = Channel.BUFFERED) { /* ... */ }
 
// UNLIMITED (capacity = Channel.UNLIMITED):
//   send 永远不挂起，内存不够时 OOM
produce(capacity = Channel.UNLIMITED) { /* ... */ }
 
// CONFLATED (capacity = Channel.CONFLATED):
//   只保留最新值，旧值被覆盖
produce(capacity = Channel.CONFLATED) { /* ... */ }

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
// 第一级管道：生成从 start 开始的自然数
fun CoroutineScope.produceNumbers(start: Int): ReceiveChannel<Int> = produce {
    var x = start                   // 从 start 开始
    while (true) {                  // 无限生成
        send(x++)                   // 发送当前值并递增
    }
}
 
// 第二级管道：过滤偶数
fun CoroutineScope.filterEven(source: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (num in source) {           // 从上游 Channel 逐个接收
        if (num % 2 == 0) {        // 只保留偶数
            send(num)               // 转发到下游
        }
    }
}
 
// 第三级管道：计算平方
fun CoroutineScope.square(source: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (num in source) {           // 从上游 Channel 逐个接收
        send(num * num)             // 发送平方值到下游
    }
}
 
fun main() = runBlocking {
    val numbers = produceNumbers(1)           // 1, 2, 3, 4, 5, ...
    val evens = filterEven(numbers)           // 2, 4, 6, 8, 10, ...
    val squares = square(evens)               // 4, 16, 36, 64, 100, ...
 
    // 只取前 5 个结果
    repeat(5) {
        println(squares.receive())            // 从最终管道接收数据
    }
 
    // 取消整个管道：cancel 最下游即可，取消会沿链路向上传播
    coroutineContext.cancelChildren()
}

4
16
36
64
100

fun main() = runBlocking {
    val numbers = produce {
        var i = 0
        while (true) {
            send(i++)               // 不断发送
            delay(100)              // 每 100ms 发送一次
        }
    }
 
    delay(550)                       // 等待约 550ms
    numbers.cancel()                 // 直接取消 ReceiveChannel
    // 此时 produce 内部的协程也会被取消
    println("管道已取消")
}

val resultChannel = produce<String>(capacity = 10) {
    // 1. 直接调用 send()（来自 SendChannel）
    send("直接发送")
 
    // 2. 启动子协程来并发生产（来自 CoroutineScope）
    launch {
        send("来自子协程 A")
    }
    launch {
        send("来自子协程 B")
    }
 
    // 3. 访问 channel 属性（ProducerScope 自身提供）
    println("Channel 是否已关闭: ${channel.isClosedForSend}")
}

val ch = produce<Int> { send(1) }
// ch 的类型是 ReceiveChannel<Int>，没有 send 方法
// ch.send(2)  // ❌ 编译错误！

val ch = produce {     // RENDEZVOUS 模式
    send(1)            // 如果没人 receive，这里将永远挂起
}
// 如果你忘记 receive，生产者就泄漏了

fun main() = runBlocking {
    val channel = produce {
        send(1)
        send(2)
        throw IllegalStateException("boom")
        send(3) // unreachable
    }
 
    try {
        for (value in channel) {
            println("Got $value")
        }
    } catch (e: Exception) {
        println("Error: ${e.message}")
    }
}

// actor 构建器的函数签名
@ObsoleteCoroutinesApi // 注意：当前标记为实验性/过时 API
fun <E> CoroutineScope.actor(
    context: CoroutineContext = EmptyCoroutineContext,  // 协程上下文，可指定调度器等
    capacity: Int = 0,                                   // Channel 缓冲容量，默认为 RENDEZVOUS（无缓冲）
    start: CoroutineStart = CoroutineStart.DEFAULT,      // 启动模式，默认立即调度
    onCompletion: CompletionHandler? = null,              // 协程完成时的回调（可选）
    block: suspend ActorScope<E>.() -> Unit              // Actor 协程体，在 ActorScope 中执行
): SendChannel<E>  // 返回值：一个 SendChannel，外部通过它向 Actor 发送消息

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
// 定义消息类型：使用密封类来描述 Actor 能处理的所有消息
sealed class CounterMsg                             // 密封类作为消息协议的基类
 
object IncCounter : CounterMsg()                    // 消息1：递增计数器（无需携带数据）
 
class GetCounter(                                   // 消息2：获取当前计数值
    val response: CompletableDeferred<Int>           // 携带一个 Deferred，Actor 会把结果写入其中
) : CounterMsg()
 
// 创建一个 counterActor，它内部维护私有计数器状态
@OptIn(ObsoleteCoroutinesApi::class)                // 显式声明使用过时 API
fun CoroutineScope.counterActor() = actor<CounterMsg> {
    var counter = 0                                  // Actor 的私有状态，外界无法直接访问
    for (msg in channel) {                           // 持续从 channel 中接收消息（挂起式循环）
        when (msg) {                                 // 根据消息类型分发处理
            is IncCounter -> counter++               // 收到递增消息 → 计数器 +1
            is GetCounter -> msg.response.complete(counter) // 收到查询消息 → 将当前值写入 response
        }
    }
}
 
fun main() = runBlocking<Unit> {
    val counter = counterActor()                     // 启动 Actor，拿到 SendChannel<CounterMsg>
 
    // 模拟大量并发递增操作
    val jobs = List(100) {                           // 创建 100 个协程
        launch {
            repeat(1000) {                           // 每个协程发送 1000 次递增消息
                counter.send(IncCounter)             // 通过 SendChannel 向 Actor 发送消息
            }
        }
    }
    jobs.forEach { it.join() }                       // 等待所有协程完成
 
    // 查询最终计数值
    val response = CompletableDeferred<Int>()        // 创建一个空的 Deferred 容器
    counter.send(GetCounter(response))               // 发送查询消息，并附带 Deferred
    println("Counter = ${response.await()}")         // 挂起等待 Actor 回写结果 → 输出: Counter = 100000
    counter.close()                                  // 关闭 Actor 的 SendChannel，Actor 协程会自然结束
}

// 只通知 Actor 做某事，不关心结果
sealed class Command                                 // 命令基类
object Increment : Command()                         // 递增
object Reset : Command()                             // 重置

// 需要 Actor 返回结果，通过 CompletableDeferred 回传
class GetValue(
    val response: CompletableDeferred<Int>           // 调用方传入一个 Deferred 作为"回信信封"
) : Command()
 
// 使用方式：
suspend fun getValue(actor: SendChannel<Command>): Int {
    val deferred = CompletableDeferred<Int>()        // 创建回信信封
    actor.send(GetValue(deferred))                   // 将信封随消息一起发给 Actor
    return deferred.await()                          // 挂起等待 Actor 写入结果
}

// 消息本身携带业务数据
data class AddItem(val item: String) : Command()     // 添加元素，携带具体的 item 数据
data class RemoveItem(val item: String) : Command()  // 移除元素

┌──────────────────────────────────────────────────────────────────┐
│                        共享内存模型（传统）                        │
│                                                                  │
│   Thread-1 ──┐                                                   │
│   Thread-2 ──┼──→  var counter = 0  ← 需要 synchronized/Mutex   │
│   Thread-3 ──┘     （共享可变状态）                                │
│                                                                  │
│   问题: 多线程直接竞争同一块内存，必须加锁                          │
├──────────────────────────────────────────────────────────────────┤
│                        Actor 消息模型                             │
│                                                                  │
│   Coroutine-1 ──┐                        ┌──────────────────┐   │
│   Coroutine-2 ──┼──→ [ SendChannel ] ──→ │ Actor 协程       │   │
│   Coroutine-3 ──┘    （消息队列 FIFO）    │ var counter = 0  │   │
│                                           │ （私有，单线程访问）│   │
│                                           └──────────────────┘   │
│   优势: 状态被 Actor 独占，消息串行处理，无需任何锁                  │
└──────────────────────────────────────────────────────────────────┘

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
 
fun CoroutineScope.counterActorManual(): SendChannel<CounterMsg> {
    val channel = Channel<CounterMsg>()              // 手动创建一个 Channel
    launch {                                         // 启动一个协程作为 Actor
        var counter = 0                              // Actor 的私有状态
        for (msg in channel) {                       // 从 channel 中持续接收消息
            when (msg) {
                is IncCounter -> counter++           // 处理递增
                is GetCounter -> msg.response.complete(counter) // 回写查询结果
            }
        }
    }
    return channel                                   // 对外返回 SendChannel
}

// 完整的生命周期演示
val channel = Channel<Int>(capacity = 3) // 创建一个容量为 3 的缓冲 Channel
 
// 生产者协程
launch {
    for (i in 1..5) {
        channel.send(i)                   // 逐个发送 1~5，缓冲满时挂起等待
        println("已发送: $i")
    }
    channel.close()                       // 发送完毕，关闭发送端（缓冲区数据不丢）
    println("Channel 已关闭")
}
 
// 消费者协程
launch {
    for (element in channel) {            // for-in 自动在 close 后退出循环
        println("已接收: $element")
        delay(200)                        // 模拟消费耗时
    }
    println("所有元素已消费完毕")           // Channel 关闭 + 缓冲区清空后到达此处
}

// produce 将「创建 Channel + 启动生产协程 + 自动关闭」三合一
val numbers: ReceiveChannel<Int> = coroutineScope.produce {
    for (i in 1..10) {
        send(i)        // 在 ProducerScope 内部直接 send
    }
    // lambda 结束，Channel 自动 close
}

// actor 将「创建 Channel + 启动消费协程」合二为一
val counterActor: SendChannel<CounterMsg> = coroutineScope.actor {
    var count = 0                     // 状态被单协程独占，无需加锁
    for (msg in channel) {            // 串行处理每条消息
        when (msg) {
            is Increment -> count++
            is GetCount  -> msg.response.complete(count)
        }
    }
}

fun main() = runBlocking {
    val channel = Channel<Int>(capacity = Channel.CONFLATED)
 
    launch {
        channel.send(1)
        channel.send(2)
        channel.send(3)
    }
 
    delay(100) // 确保生产者先执行完
 
    val value = channel.receive()
    println(value)
}