Channel 与 BroadcastChannel（Channel and BroadcastChannel）

前置知识#

• Flow 冷流与 SharedFlow 和 StateFlow：建议先完成前一篇的学习

学习目标#

• 掌握「1. 历史动机与发展脉络」的核心机制、典型用法与常见陷阱
• 掌握「2. 形式化定义」的核心机制、典型用法与常见陷阱
• 掌握「3. 理论推导与原理解析」的核心机制、典型用法与常见陷阱
• 掌握「4. 代码示例」的核心机制、典型用法与常见陷阱
• 掌握「5. 对比分析」的核心机制、典型用法与常见陷阱

本文档对标 MIT 6.005、Stanford CS193P、CMU 15-410 教学水准，系统讲解 Kotlin 协程中的 Channel<T>、SendChannel/ReceiveChannel、BroadcastChannel（已废弃）、SharedFlow 替代方案、produce/actor 构建器以及 select 多路复用机制。内容覆盖 CSP（Communicating Sequential Processes）理论基础、Kotlin 1.0 至 2.0 的演进、JVM 字节码实现、跨语言对比、企业级生产代码与习题解析。

1. 历史动机与发展脉络#

1.1 问题背景：协程间通信的挑战#

协程的核心价值在于”以同步写法表达异步逻辑”，但多个协程之间如何安全地共享数据、传递消息，是一个长期存在的工程难题。传统的并发通信方案可分为三类：

方案一：共享可变状态 + 锁（Shared Mutable State + Locks）

• 优势：性能高（无序列化开销）、内存占用低。
• 劣势：易出现死锁、竞态条件、可重入问题；难以推理程序行为。
• 典型场景：Java 的 synchronized / ReentrantLock、ConcurrentHashMap。

方案二：Actor 模型（Actor Model）

• 优势：天然解耦、消息顺序保证、无共享状态。
• 劣势：消息序列化开销、错误传播复杂、调试困难。
• 典型场景：Akka（Scala/Erlang）、Swift Actor。

方案三：CSP 模型（Communicating Sequential Processes）

• 优势：类型安全的消息传递、编译期检查、背压支持。
• 劣势：仅支持单消费者（默认）、缓冲策略选择复杂。
• 典型场景：Go channel、Kotlin Channel、Rust mpsc。

Kotlin 协程选择了 CSP 模型 作为协程间通信的核心原语，原因：

1. 与 suspend 函数契合：send/receive 作为挂起函数，天然支持非阻塞等待。
2. 结构化并发友好：Channel 的生命周期可绑定到 CoroutineScope，避免泄漏。
3. 类型安全：Channel<T> 通过泛型保证消息类型一致，编译期检查。
4. 与 Flow 生态统一：Channel 可与 Flow 互转，形成统一的数据流抽象。

1.2 学术背景：CSP 与 Go channel#

CSP（Communicating Sequential Processes）由 Tony Hoare 于 1978 年提出，是并发计算的数学模型。其核心思想：

Go 语言于 2009 年首次发布，将 CSP 模型工业化为 chan 类型，证明”CSP 是可行的工业级并发原语”。Go channel 的设计要点：

1. 会合语义（Rendezvous）：无缓冲 channel 的发送方与接收方必须同步会合。
2. FIFO 顺序：消息按发送顺序到达接收方。
3. 关闭传播：关闭 channel 后，所有阻塞的接收方收到零值。
4. select 多路复用：在多个 channel 操作中选择就绪的一个。

Kotlin 的 Channel 借鉴了 Go channel 的设计，但做了若干改进：

1.3 Kotlin 1.0（2016 年）：无 Channel 时代#

Kotlin 1.0 不支持协程，开发者使用传统的并发原语：

// Kotlin 1.0 - 共享可变状态 + 锁
class Counter {
    private val lock = ReentrantLock()
    private var count = 0
    
    fun increment() {
        lock.withLock {
            count++
        }
    }
}

这种方式的痛点：

• 死锁风险高。
• 难以调试。
• 缺乏类型安全保障。

1.4 Kotlin 1.1（2017 年 5 月）：协程实验性引入 Channel#

Kotlin 1.1 引入协程作为实验性特性，同时引入 Channel 作为协程间通信原语：

// Kotlin 1.1
val channel = Channel<Int>()

launch {
    channel.send(1)
}

launch {
    val value = channel.receive()
    println(value)  // 1
}

此时的 Channel 已具备完整的核心 API：

• Channel<T>()：创建实例。
• send(value: T)：挂起发送。
• receive(): T：挂起接收。
• close()：关闭发送端。
• produce { ... }：生产者构建器。
• actor<T> { ... }：Actor 构建器（实验性）。

1.5 Kotlin 1.2-1.3（2017-2018）：API 完善与 GA#

Kotlin 1.2 引入 BroadcastChannel 作为多播方案：

val broadcast = BroadcastChannel<Int>(bufferCapacity = 10)

launch {
    broadcast.send(1)
}

launch {
    val subscriber = broadcast.openSubscription()
    subscriber.receive()  // 1
}

Kotlin 1.3 将协程提升为 GA，Channel API 稳定化：

1. @ExperimentalCoroutinesApi：标记实验性 API。
2. conflate() 操作符：将任意 Channel 转换为 CONFLATED 模式。
3. select 表达式 GA：支持多路复用。
4. consumeEach 扩展：安全的遍历消费。

1.6 Kotlin 1.4-1.5（2020-2021）：BroadcastChannel 走向废弃#

随着 Flow API 的成熟，BroadcastChannel 的局限性逐渐显现：

1. 缺少 replay 缓冲：新订阅者无法获取历史值。
2. 缺少 BufferOverflow 策略：仅支持丢弃最新或阻塞。
3. 与 Flow 生态不统一：无法直接转换为 Flow。

Kotlin 1.4 引入 SharedFlow 作为替代方案，1.5 将 BroadcastChannel 标记为 @ObsoleteCoroutinesApi：

// 旧方式（已废弃）
val broadcast = BroadcastChannel<Int>(10)

// 新方式：使用 MutableSharedFlow
val sharedFlow = MutableSharedFlow<Int>(
    replay = 0,
    extraBufferCapacity = 10,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)

1.7 Kotlin 1.6-1.7（2021-2022）：actor 废弃与 select 改进#

Kotlin 1.6 将 actor 构建器标记为废弃，推荐使用 Channel + 状态封装协程：

// 旧方式（已废弃）
val counterActor = actor<Int> {
    var count = 0
    for (msg in channel) {
        when (msg) {
            is Increment -> count++
            is Get -> reply(count)
        }
    }
}

// 新方式：使用 Channel + launch
fun CoroutineScope.counterActor(): Channel<CounterMsg> {
    val channel = Channel<CounterMsg>()
    var count = 0
    launch {
        for (msg in channel) {
            when (msg) {
                is Increment -> count++
                is Get -> msg.reply.complete(count)
            }
        }
    }
    return channel
}

Kotlin 1.7 对 select 进行优化：

1. 更少的对象分配：K2 编译器减少了 SelectInstance 的中间对象。
2. 更精确的类型推断：分支返回类型自动收窄。
3. 更好的诊断：未处理的 clause 会有警告。

1.8 Kotlin 1.8-1.9（2023 年）：Channel 与 Flow 桥接完善#

Kotlin 1.8 引入 receiveAsFlow() 与 consumeAsFlow() 扩展，将 Channel 转换为 Flow：

val channel = Channel<Int>()
val flow: Flow<Int> = channel.receiveAsFlow()

// 消费者侧
flow.collect { value ->
    println(value)
}

Kotlin 1.9 进一步引入 Channel(capacity = Channel.RENDEZVOUS) 的实验性变体 Channel(Channel.RENDEZVOUS, onBufferOverflow = ...)，允许为 RENDEZVOUS 模式配置溢出策略（虽然 RENDEZVOUS 模式下缓冲始终为 0，溢出策略仅影响 trySend 的行为）。

1.9 Kotlin 2.0（2024 年 5 月）：K2 优化#

Kotlin 2.0 的 K2 编译器对 Channel 与 select 进行了内部优化：

1. 状态机对象复用：send/receive 编译为状态机后，K2 能更好地复用 Continuation 对象，减少 GC 压力。
2. select 内联优化：K2 将简单的 select 表达式内联为 if-else 形式，减少运行时开销。
3. LockFreeLinkedList 优化：JVM 平台的 CAS 操作进一步优化，吞吐量提升约 15%。
4. KMP 一致性：JVM、JS、Native、Wasm 平台的 Channel 行为完全一致。

1.10 JetBrains 的设计哲学#

JetBrains 在设计 Channel 时遵循了以下哲学：

1. CSP 优先：协程间通信首选消息传递而非共享状态。
2. 挂起优于阻塞：send/receive 是挂起函数，不阻塞线程。
3. 接口分离：SendChannel 与 ReceiveChannel 独立，支持最小权限。
4. 显式关闭：Channel 必须显式 close()，避免资源泄漏。
5. 背压天然支持：缓冲满后 send 挂起，无需额外背压机制。
6. 与 Flow 生态统一：Channel 可双向转换为 Flow，形成统一抽象。
7. 废弃不删除：BroadcastChannel/actor 标记废弃但保留，向后兼容。

1.11 时间线总览#

2016  Kotlin 1.0 — 无协程，使用传统并发原语
2017  Kotlin 1.1 — 协程实验性，Channel 首发
2018  Kotlin 1.3 — 协程 GA，Channel API 稳定
2020  Kotlin 1.4 — SharedFlow 引入
2021  Kotlin 1.5 — BroadcastChannel 标记废弃
2022  Kotlin 1.7 — actor 废弃，select 优化
2023  Kotlin 1.9 — Channel-Flow 桥接完善
2024  Kotlin 2.0 — K2 优化，KMP 一致

2. 形式化定义#

2.1 Channel 的类型定义#

根据 Kotlin 官方文档，Channel<T> 的形式化定义如下：

其中：

• $\text{SendChannel}\langle T \rangle$ 提供 send、trySend、close、isClosedForSend。
• $\text{ReceiveChannel}\langle T \rangle$ 提供 receive、tryReceive、receiveCatching、cancel、isClosedForReceive。

Channel<T> 是这两个接口的合取，既是发送端也是接收端。若仅需发送权限，可声明为 SendChannel<T>；若仅需接收权限，可声明为 ReceiveChannel<T>。

2.2 容量模式的形式化#

Channel 的容量（capacity）形式化定义：

其中：

• $\text{RENDEZVOUS} = 0$：无缓冲，发送方必须等待接收方就绪。
• $\text{UNLIMITED} = \infty$：无上限缓冲，发送方永不阻塞（内存耗尽风险）。
• $\text{BUFFERED}(n)$：固定容量 $n$，默认 $n = 64$。
• $\text{CONFLATED} = 1$（仅保留最新值）：新值覆盖旧值。

形式化地，send 操作的语义：

2.3 会合语义（Rendezvous）#

RENDEZVOUS 模式下，发送方与接收方的会合点（Rendezvous Point）可形式化为：

即发送方 $s$ 与接收方 $r$ 必须同时就绪，才能完成值的传递。这种同步语义确保：

1. 零缓冲：没有内存压力。
2. 强同步：发送方知道接收方已收到。
3. 强背压：发送速率被接收速率限制。

2.4 FIFO 顺序保证#

Channel 保证 FIFO（First-In-First-Out）顺序：

即：若 $v_i$ 在 $v_j$ 之前被 send，则 $v_i$ 在 $v_j$ 之前被 receive。

形式化地，Channel 是一个 FIFO 队列：

2.5 关闭语义的形式化#

close() 操作的形式化语义：

调用后：

1. 发送端关闭：isClosedForSend = true，后续 send 抛出 ClosedSendChannelException。
2. 接收端保持开启：直到缓冲区清空后，isClosedForReceive = true，receive 抛出 ClosedReceiveChannelException。
3. receiveCatching：返回 ChannelResult.closed(cause)，不抛异常。

形式化地，关闭后的状态转换：

2.6 取消语义的形式化#

cancel() 操作的形式化语义：

调用后：

1. 立即关闭：isClosedForReceive = true。
2. 缓冲丢弃：缓冲区中的所有元素被丢弃。
3. 等待发送方唤醒：所有阻塞在 send 上的协程被唤醒并抛出 CancellationException。

形式化地，取消操作的状态转换：

2.7 BroadcastChannel 的形式化（已废弃）#

BroadcastChannel<T> 的形式化定义：

其中：

• $\text{BroadcastSender}$ 提供 send、close。
• $\text{BroadcastSubscriberFactory}$ 提供 openSubscription(): ReceiveChannel<T>，每次调用返回独立的订阅通道。

形式化地，广播语义：

即：每个订阅者独立接收消息副本，互不影响。但这种设计的局限：

1. 无 replay：新订阅者无法获取历史值。
2. 订阅者有界：超过 subscribers 上限后新订阅失败。
3. 资源开销大：每个订阅者维护独立的 ReceiveChannel。

2.8 SharedFlow 的形式化（替代方案）#

SharedFlow<T> 的形式化定义：

核心特性：

1. replay 缓冲：新订阅者获取最近 $n$ 个值（replay = n）。
2. extraBufferCapacity：额外的缓冲容量，用于背压控制。
3. onBufferOverflow：缓冲满时的策略（SUSPEND/DROP_OLDEST/DROP_LATEST）。

形式化地，订阅语义：

即：先回放历史值，再转发新值。

2.9 select 表达式的形式化#

select 表达式形式化定义：

执行语义：

若多个 clause 同时就绪，随机选择一个（避免饥饿）。

形式化地，select 是一种多路复用（Multiplexing）：

3. 理论推导与原理解析#

3.1 无锁队列的实现原理#

Channel 的核心数据结构是 LockFreeLinkedListHead，一种基于 CAS（Compare-And-Swap）的无锁链表：

// 简化的无锁链表节点
class LockFreeLinkedListNode {
    val next: AtomicReference<LockFreeLinkedListNode?> = AtomicReference(null)
    val value: Any? = null
    
    fun addLast(value: Any?): Boolean {
        val newNode = LockFreeLinkedListNode().apply { this.value = value }
        // CAS 操作：尝试将 next 从 null 更新为 newNode
        return next.compareAndSet(null, newNode)
    }
    
    fun removeFirst(): Any? {
        while (true) {
            val first = next.get()
            if (first == null) return null
            // CAS 操作：尝试将 next 从 first 更新为 first.next
            if (next.compareAndSet(first, first.next.get())) {
                return first.value
            }
            // CAS 失败，重试
        }
    }
}

形式化地，CAS 操作的语义：

CAS 是原子操作，由 CPU 指令直接支持（x86 的 lock cmpxchg、ARM 的 ldrex/strex）。

3.2 会合点的同步机制#

RENDEZVOUS 模式下，send 与 receive 必须会合。会合点的实现机制：

1. send 先到：将值放入”待发送”槽，挂起等待接收方。
2. receive 先到：将”待接收”槽标记，挂起等待发送方。
3. 会合完成：发送方与接收方都就绪，值被传递，双方恢复。

形式化地，会合点的状态机：

flowchart TD
    C0_0["send(v) arrives"]
    C0_1["Empty   Waiting Done"]
    C0_2["S: Empty (无等待)"]
    C0_3["W: Waiting (send 等待 receive)"]
    C0_4["D: Done (会合完成)"]
    C1_0["S"]
    C2_0[">"]
    C3_0["W"]
    C4_0[">"]
    C5_0["D"]
    C0_0 --> C0_1
    C0_1 --> C0_2
    C0_2 --> C0_3
    C0_3 --> C0_4
    C0_0 --> C1_0
    C1_0 --> C2_0
    C2_0 --> C3_0
    C3_0 --> C4_0
    C4_0 --> C5_0

在 kotlinx.coroutines 源码中，会合点通过 AbstractSendChannel 的 BufferedChannel 实现：

// 简化的会合点逻辑
suspend fun send(value: T) {
    // 1. 尝试立即交付给等待的接收方
    if (tryOfferToReceiver(value)) return
    
    // 2. 尝试放入缓冲区
    if (tryAddToBuffer(value)) return
    
    // 3. 挂起等待接收方
    return suspendCancellableCoroutine { cont ->
        val sendElement = SendElement(value, cont)
        queue.addLast(sendElement)
    }
}

suspend fun receive(): T {
    // 1. 尝试从缓冲区取出
    buffer.poll()?.let { return it }
    
    // 2. 尝试从等待的发送方取出
    queue.poll()?.let { sendElement ->
        sendElement.cont.resume(sendElement.value)
        return sendElement.value
    }
    
    // 3. 挂起等待发送方
    return suspendCancellableCoroutine { cont ->
        val receiveElement = ReceiveElement(cont)
        queue.addLast(receiveElement)
    }
}

3.3 CONFLATED 模式的实现#

CONFLATED 模式下，缓冲区仅保留最新值。实现机制：

// 简化的 CONFLATED 实现
class ConflatedChannel<T> : Channel<T> {
    private val value = AtomicReference<Any?>(UNSET)
    
    override suspend fun send(value: T) {
        this.value.set(value)  // 直接覆盖
    }
    
    override suspend fun receive(): T {
        while (true) {
            val v = value.get()
            if (v != UNSET) {
                if (value.compareAndSet(v, UNSET)) {
                    return v as T
                }
            } else {
                // 挂起等待新值
                suspendCancellableCoroutine { cont ->
                    // 注册监听器，新值到达时恢复
                }
            }
        }
    }
}

形式化地，CONFLATED 的语义：

即：每次 send 都覆盖缓冲区，永远只保留最新值。

3.4 BUFFERED 模式的实现#

BUFFERED 模式下，使用环形缓冲区（Ring Buffer）：

// 简化的环形缓冲区
class RingBuffer<T>(val capacity: Int) {
    private val buffer = arrayOfNulls<Any>(capacity)
    private val head = AtomicInteger(0)
    private val tail = AtomicInteger(0)
    
    fun offer(value: T): Boolean {
        while (true) {
            val currentTail = tail.get()
            val currentHead = head.get()
            if (currentTail - currentHead >= capacity) {
                return false  // 缓冲满
            }
            if (tail.compareAndSet(currentTail, currentTail + 1)) {
                buffer[currentTail % capacity] = value
                return true
            }
        }
    }
    
    fun poll(): T? {
        while (true) {
            val currentHead = head.get()
            val currentTail = tail.get()
            if (currentHead >= currentTail) {
                return null  // 缓冲空
            }
            val value = buffer[currentHead % capacity]
            if (head.compareAndSet(currentHead, currentHead + 1)) {
                @Suppress("UNCHECKED_CAST")
                return value as T
            }
        }
    }
}

形式化地，环形缓冲区的索引：

这种设计避免了链表节点的动态分配，提升性能。

3.5 select 表达式的实现原理#

select 表达式的核心是 SelectInstance，它聚合多个 SelectClause：

// 简化的 select 实现
class SelectInstance<R> {
    private val selected = AtomicReference<ClauseResult<R>?>(null)
    private val clauses = mutableListOf<SelectClause<R>>()
    
    fun select(): R {
        // 1. 注册所有 clause
        clauses.forEach { clause ->
            clause.register(this) { result ->
                // clause 就绪，尝试设置选中结果
                if (selected.compareAndSet(null, result)) {
                    // 取消其他 clause
                    clauses.forEach { other ->
                        if (other !== clause) other.cancel()
                    }
                }
            }
        }
        
        // 2. 挂起等待
        return suspendCoroutine { cont ->
            // 监听 selected 的变化
        }
    }
}

形式化地，select 的执行流程：

即：等待第一个 clause 就绪，取消其他 clause。

3.6 关闭与取消的传播#

Channel 的关闭与取消遵循以下传播规则：

1. close() 仅关闭发送端：接收端可继续接收缓冲区中的剩余元素。
2. cancel() 关闭接收端：缓冲区中的元素被丢弃，发送方的 send 抛出 CancellationException。
3. 结构化并发绑定：若 Channel 由 produce { ... } 创建，协程取消时 Channel 自动关闭。

形式化地，状态转换：

            close()           drain buffer
Open ----------------> Closing ----------------> Closed
  |                                              ^
  |                cancel()                      |
  +--------------------------------------------->+

3.7 Channel 与 Flow 的桥接#

Channel 与 Flow 的桥接通过以下扩展函数实现：

// Channel -> Flow
fun <T> ReceiveChannel<T>.receiveAsFlow(): Flow<T> = flow {
    for (value in this@receiveAsFlow) {
        emit(value)
    }
}

// Flow -> Channel
fun <T> Flow<T>.produceIn(scope: CoroutineScope): ReceiveChannel<T> =
    scope.produce {
        collect { send(it) }
    }

形式化地，桥接的语义：

这种桥接让 Channel 与 Flow 形成统一的数据流抽象，可在两种 API 间自由切换。

3.8 JVM 字节码层面#

在 JVM 平台上，Channel 的 send/receive 编译为带 Continuation 参数的方法：

方法签名:
public Object send(T value, Continuation<? super Unit> p1)

异常表: 可选（仅当 send 内部有 try-catch）

select 表达式编译为 SelectBuilder 的实现：

public static Object select(SelectBuilder<R> builder, Continuation<? super R> p1) {
    // 注册所有 clause
    builder.apply(new SelectBuilderImpl(p1));
    // 等待结果
    return COROUTINE_SUSPENDED;
}

编译后的字节码中，select 的 clause 被编译为独立的 lambda 表达式，运行时按需执行。

4. 代码示例#

4.1 基础 Channel 使用#

// Kotlin 2.0
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

suspend fun main() {
    val channel = Channel<Int>()  // 默认 RENDEZVOUS
    
    // 启动发送方协程
    launch {
        repeat(5) { i ->
            println("Sending $i")
            channel.send(i)  // 挂起直到接收方就绪
            println("Sent $i")
        }
        channel.close()
    }
    
    // 启动接收方协程
    launch {
        for (value in channel) {
            println("Received $value")
            delay(100)  // 模拟处理耗时
        }
    }
    
    delay(2000)
}

输出（顺序可能略有差异，但发送/接收配对）：

Sending 0
Received 0
Sent 0
Sending 1
Received 1
Sent 1
...

4.2 四种容量模式对比#

// Kotlin 2.0
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

suspend fun main() = coroutineScope {
    // 模式 1: RENDEZVOUS（默认，无缓冲）
    val rendezvous = Channel<Int>()  // 等价于 Channel<Int>(Channel.RENDEZVOUS)
    
    // 模式 2: UNLIMITED（无上限缓冲）
    val unlimited = Channel<Int>(Channel.UNLIMITED)
    
    // 模式 3: BUFFERED（固定 64 缓冲）
    val buffered = Channel<Int>(Channel.BUFFERED)
    val customBuffered = Channel<Int>(10)  // 自定义 10 缓冲
    
    // 模式 4: CONFLATED（仅保留最新值）
    val conflated = Channel<Int>(Channel.CONFLATED)
    
    // 演示 RENDEZVOUS：发送方必须等待接收方
    launch {
        rendezvous.send(1)  // 会挂起直到接收方就绪
        println("RENDEZVOUS: sent 1")
    }
    launch {
        delay(100)
        val v = rendezvous.receive()
        println("RENDEZVOUS: received $v")
    }
    
    // 演示 UNLIMITED：发送方永不阻塞
    launch {
        repeat(10) { i ->
            unlimited.send(i)  // 立即返回
        }
        println("UNLIMITED: sent 10 values without blocking")
    }
    
    // 演示 BUFFERED：缓冲满后挂起
    launch {
        repeat(15) { i ->  // 15 < 64，不会阻塞
            customBuffered.send(i)
        }
        println("BUFFERED(10): sent 15 values")
    }
    
    // 演示 CONFLATED：仅保留最新值
    launch {
        conflated.send(1)
        conflated.send(2)  // 覆盖 1
        conflated.send(3)  // 覆盖 2
        println("CONFLATED: sent 1, 2, 3")
    }
    launch {
        delay(100)
        val v = conflated.receive()  // 仅能收到 3
        println("CONFLATED: received $v")
    }
    
    delay(500)
}

4.3 produce 构建器：生产者协程#

// Kotlin 2.0
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun CoroutineScope.numberProducer(start: Int, end: Int): ReceiveChannel<Int> = produce {
    for (i in start..end) {
        send(i)
        delay(100)  // 模拟生产耗时
    }
}

suspend fun main() = coroutineScope {
    val producer = numberProducer(1, 10)
    
    // 消费者
    producer.consumeEach { number ->
        println("Consumed: $number")
    }
    
    println("Done")
}

输出：

Consumed: 1
Consumed: 2
...
Consumed: 10
Done

4.4 工作池模式（Worker Pool）#

// Kotlin 2.0 - 企业级工作池实现
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import java.util.concurrent.atomic.AtomicInteger

sealed interface WorkTask {
    val id: String
    data class ProcessOrder(override val id: String, val orderId: String) : WorkTask
    data class SendEmail(override val id: String, val to: String, val subject: String) : WorkTask
    data class GenerateReport(override val id: String, val reportType: String) : WorkTask
}

class WorkResult(val taskId: String, val success: Boolean, val message: String)

class WorkerPool(
    private val workerCount: Int,
    private val capacity: Int = Channel.UNLIMITED
) {
    private val workQueue = Channel<WorkTask>(capacity)
    private val resultChannel = Channel<WorkResult>(capacity)
    private val processedCount = AtomicInteger(0)
    
    suspend fun submit(task: WorkTask) {
        workQueue.send(task)
    }
    
    fun start(scope: CoroutineScope) {
        repeat(workerCount) { workerId ->
            scope.launch {
                for (task in workQueue) {
                    try {
                        val result = processTask(task, workerId)
                        resultChannel.send(result)
                        processedCount.incrementAndGet()
                    } catch (e: Exception) {
                        resultChannel.send(WorkResult(task.id, false, "Error: ${e.message}"))
                    }
                }
            }
        }
    }
    
    private suspend fun processTask(task: WorkTask, workerId: Int): WorkResult {
        return when (task) {
            is WorkTask.ProcessOrder -> {
                delay(50)  // 模拟处理
                WorkResult(task.id, true, "Worker $workerId processed order ${task.orderId}")
            }
            is WorkTask.SendEmail -> {
                delay(100)
                WorkResult(task.id, true, "Worker $workerId sent email to ${task.to}")
            }
            is WorkTask.GenerateReport -> {
                delay(200)
                WorkResult(task.id, true, "Worker $workerId generated ${task.reportType} report")
            }
        }
    }
    
    fun results(): ReceiveChannel<WorkResult> = resultChannel
    
    suspend fun stop() {
        workQueue.close()
        workQueue.consume { }  // 等待所有任务处理完成
        resultChannel.close()
    }
    
    fun processedCount(): Int = processedCount.get()
}

suspend fun main() = coroutineScope {
    val pool = WorkerPool(workerCount = 4)
    
    // 启动工作池
    pool.start(this)
    
    // 启动结果收集器
    launch {
        for (result in pool.results()) {
            println("[Result] ${result.taskId}: ${result.message}")
        }
    }
    
    // 提交任务
    repeat(20) { i ->
        val task = when (i % 3) {
            0 -> WorkTask.ProcessOrder("task-$i", "ORD-$i")
            1 -> WorkTask.SendEmail("task-$i", "user$i@example.com", "Hello $i")
            else -> WorkTask.GenerateReport("task-$i", "Monthly")
        }
        pool.submit(task)
    }
    
    delay(3000)
    pool.stop()
    println("Total processed: ${pool.processedCount()}")
}

4.5 事件总线（基于 SharedFlow 替代 BroadcastChannel）#

// Kotlin 2.0 - 现代事件总线实现
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.channels.*

sealed interface AppEvent {
    val timestamp: Long
    data class UserLoggedIn(val userId: String, override val timestamp: Long) : AppEvent
    data class UserLoggedOut(val userId: String, override val timestamp: Long) : AppEvent
    data class DataChanged(val key: String, override val timestamp: Long) : AppEvent
    data class ErrorOccurred(val error: Throwable, override val timestamp: Long) : AppEvent
}

class EventBus {
    // 使用 MutableSharedFlow 替代废弃的 BroadcastChannel
    private val _events = MutableSharedFlow<AppEvent>(
        replay = 0,                    // 新订阅者不获取历史值
        extraBufferCapacity = 100,     // 额外缓冲 100 个事件
        onBufferOverflow = BufferOverflow.DROP_OLDEST  // 满时丢弃最旧
    )
    
    val events: SharedFlow<AppEvent> = _events.asSharedFlow()
    
    suspend fun publish(event: AppEvent) {
        _events.emit(event)
    }
    
    // 按事件类型过滤订阅
    inline fun <reified T : AppEvent> subscribeFiltered(): Flow<T> {
        return events.filter { it is T }.map { it as T }
    }
}

class AuthService(private val eventBus: EventBus) {
    suspend fun login(userId: String, password: String): Boolean {
        delay(100)  // 模拟认证
        val success = password == "valid"
        if (success) {
            eventBus.publish(AppEvent.UserLoggedIn(userId, System.currentTimeMillis()))
        }
        return success
    }
    
    suspend fun logout(userId: String) {
        eventBus.publish(AppEvent.UserLoggedOut(userId, System.currentTimeMillis()))
    }
}

class LoggingService(private val eventBus: EventBus) {
    suspend fun startLogging(scope: CoroutineScope) {
        eventBus.events.collect { event ->
            println("[LOG] ${event.timestamp}: $event")
        }
    }
}

class AnalyticsService(private val eventBus: EventBus) {
    suspend fun startAnalytics(scope: CoroutineScope) {
        eventBus.subscribeFiltered<AppEvent.UserLoggedIn>().collect { event ->
            println("[ANALYTICS] User logged in: ${event.userId}")
        }
    }
}

suspend fun main() = coroutineScope {
    val eventBus = EventBus()
    val authService = AuthService(eventBus)
    
    // 启动日志服务
    launch { LoggingService(eventBus).startLogging(this) }
    
    // 启动分析服务
    launch { AnalyticsService(eventBus).startAnalytics(this) }
    
    delay(100)  // 等待订阅者就绪
    
    // 发布事件
    authService.login("user-001", "valid")
    delay(50)
    authService.logout("user-001")
    
    delay(500)
}

4.6 select 多路复用#

// Kotlin 2.0 - select 多路复用示例
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.selects.*

suspend fun main() = coroutineScope {
    val channel1 = Channel<String>()
    val channel2 = Channel<String>()
    val channel3 = Channel<String>()
    
    // 启动三个生产者
    launch {
        delay(100)
        channel1.send("from channel1")
    }
    launch {
        delay(50)
        channel2.send("from channel2")
    }
    launch {
        delay(150)
        channel3.send("from channel3")
    }
    
    // 使用 select 等待第一个就绪的 channel
    repeat(3) {
        val result = select<String> {
            channel1.onReceive { it }
            channel2.onReceive { it }
            channel3.onReceive { it }
        }
        println("Received: $result")
    }
    
    // select 带超时
    val timeoutResult = select<String?> {
        channel1.onReceive { it }
        channel2.onReceive { it }
        onTimeout(100) { null }  // 100ms 超时返回 null
    }
    println("Timeout result: $timeoutResult")
    
    // select 用于发送
    val outChannel1 = Channel<Int>()
    val outChannel2 = Channel<Int>()
    
    launch {
        delay(100)
        outChannel1.receive()
        println("outChannel1 received")
    }
    launch {
        delay(50)
        outChannel2.receive()
        println("outChannel2 received")
    }
    
    select<Unit> {
        outChannel1.onSend(42) { }
        outChannel2.onSend(42) { }
    }
}

4.7 管道（Pipeline）模式#

// Kotlin 2.0 - 流式处理管道
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun CoroutineScope.numbers(start: Int, count: Int): ReceiveChannel<Int> = produce {
    repeat(count) { i ->
        send(start + i)
        delay(50)
    }
}

fun CoroutineScope.square(source: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (x in source) {
        send(x * x)
    }
}

fun CoroutineScope.filter(source: ReceiveChannel<Int>, predicate: (Int) -> Boolean): ReceiveChannel<Int> = produce {
    for (x in source) {
        if (predicate(x)) {
            send(x)
        }
    }
}

fun CoroutineScope.toStrings(source: ReceiveChannel<Int>): ReceiveChannel<String> = produce {
    for (x in source) {
        send("Number: $x")
    }
}

suspend fun main() = coroutineScope {
    // 构建管道：numbers -> square -> filter(偶数) -> toStrings
    val source = numbers(1, 20)
    val squared = square(source)
    val evenSquared = filter(squared) { it % 2 == 0 }
    val strings = toStrings(evenSquared)
    
    // 消费最终结果
    for (s in strings) {
        println(s)
    }
    
    println("Pipeline complete")
}

输出：

Number: 4
Number: 16
Number: 36
Number: 64
Number: 100
...
Pipeline complete

4.8 自定义 Channel 配置#

// Kotlin 2.0 - 自定义 Channel 配置
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*

suspend fun main() = coroutineScope {
    // 自定义容量与溢出策略
    val channel = Channel<Int>(
        capacity = 10,
        onBufferOverflow = BufferOverflow.SUSPEND,  // 满时挂起（默认）
        onUndeliveredElement = { element, _ ->
            println("Element $element was not delivered (channel closed)")
        }
    )
    
    // drop-oldest 策略
    val dropOldest = Channel<Int>(
        capacity = 3,
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    )
    
    launch {
        repeat(10) { i ->
            val result = dropOldest.trySend(i)
            println("trySend($i): $result")
        }
        dropOldest.close()
    }
    
    launch {
        delay(500)  // 让发送方先发完
        for (v in dropOldest) {
            println("Received: $v")
        }
    }
    
    delay(1000)
}

4.9 Channel 与 Flow 互转#

// Kotlin 2.0
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*

suspend fun main() = coroutineScope {
    // Channel -> Flow
    val channel = Channel<Int>()
    val flow: Flow<Int> = channel.receiveAsFlow()
    
    launch {
        channel.send(1)
        channel.send(2)
        channel.send(3)
        channel.close()
    }
    
    flow.collect { value ->
        println("From Flow: $value")
    }
    
    // Flow -> Channel
    val sourceFlow = flowOf("a", "b", "c")
    val resultChannel = sourceFlow.produceIn(this)
    
    for (s in resultChannel) {
        println("From Channel: $s")
    }
    
    // consumeAsFlow：单次消费的 Flow
    val eventChannel = Channel<String>()
    launch {
        repeat(3) { i ->
            eventChannel.send("event-$i")
        }
        eventChannel.close()
    }
    
    eventChannel.consumeAsFlow().collect { event ->
        println("Event: $event")
    }
}

4.10 完整的消费者-生产者系统#

// Kotlin 2.0 - 企业级生产者-消费者系统
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*
import java.util.concurrent.atomic.AtomicLong

// 业务数据模型
data class Order(
    val id: String,
    val customerId: String,
    val items: List<String>,
    val totalAmount: Double,
    val createdAt: Long
)

data class ProcessedOrder(
    val order: Order,
    val status: OrderStatus,
    val processedAt: Long,
    val note: String = ""
)

enum class OrderStatus { SUCCESS, FAILED, RETRY }

// 订单生产者
class OrderProducer(
    private val output: SendChannel<Order>,
    private val totalOrders: Int
) {
    suspend fun startProducing() {
        repeat(totalOrders) { i ->
            val order = Order(
                id = "ORD-${System.currentTimeMillis()}-$i",
                customerId = "CUST-$i",
                items = listOf("item-1", "item-2", "item-3"),
                totalAmount = 100.0 * (i + 1),
                createdAt = System.currentTimeMillis()
            )
            output.send(order)
            delay(50)  // 模拟订单到达间隔
        }
        output.close()
    }
}

// 订单处理器
class OrderProcessor(
    private val input: ReceiveChannel<Order>,
    private val output: SendChannel<ProcessedOrder>,
    private val workerId: String
) {
    private val processedCount = AtomicLong(0)
    
    suspend fun startProcessing() {
        for (order in input) {
            val processed = process(order)
            output.send(processed)
            processedCount.incrementAndGet()
        }
    }
    
    private suspend fun process(order: Order): ProcessedOrder {
        return try {
            delay(100)  // 模拟处理
            ProcessedOrder(order, OrderStatus.SUCCESS, System.currentTimeMillis())
        } catch (e: Exception) {
            ProcessedOrder(order, OrderStatus.FAILED, System.currentTimeMillis(), e.message ?: "Unknown error")
        }
    }
    
    fun processedCount(): Long = processedCount.get()
}

// 结果收集器
class ResultCollector(private val input: ReceiveChannel<ProcessedOrder>) {
    private val successCount = AtomicLong(0)
    private val failedCount = AtomicLong(0)
    
    suspend fun startCollecting() {
        for (processed in input) {
            when (processed.status) {
                OrderStatus.SUCCESS -> {
                    successCount.incrementAndGet()
                    println("[SUCCESS] ${processed.order.id}")
                }
                OrderStatus.FAILED -> {
                    failedCount.incrementAndGet()
                    println("[FAILED] ${processed.order.id}: ${processed.note}")
                }
                OrderStatus.RETRY -> {
                    println("[RETRY] ${processed.order.id}")
                }
            }
        }
    }
    
    fun getStats(): Pair<Long, Long> = successCount.get() to failedCount.get()
}

suspend fun main() = coroutineScope {
    val orderChannel = Channel<Order>(capacity = 100)
    val resultChannel = Channel<ProcessedOrder>(capacity = 100)
    
    val producer = OrderProducer(orderChannel, totalOrders = 50)
    val processors = (1..3).map {
        OrderProcessor(orderChannel, resultChannel, "worker-$it")
    }
    val collector = ResultCollector(resultChannel)
    
    val startTime = System.currentTimeMillis()
    
    // 启动所有组件
    val producerJob = launch { producer.startProducing() }
    val processorJobs = processors.map { p ->
        launch { p.startProcessing() }
    }
    val collectorJob = launch { collector.startCollecting() }
    
    // 等待所有任务完成
    producerJob.join()
    processorJobs.forEach { it.join() }
    resultChannel.close()
    collectorJob.join()
    
    val duration = System.currentTimeMillis() - startTime
    val (success, failed) = collector.getStats()
    
    println("\n===== Final Statistics =====")
    println("Total duration: ${duration}ms")
    println("Success: $success")
    println("Failed: $failed")
    processors.forEachIndexed { i, p ->
        println("Worker-${i + 1} processed: ${p.processedCount()}")
    }
}

4.11 跨平台 Channel 使用#

// Kotlin 2.0 - KMP 项目中的 Channel
// commonMain
expect fun currentTimeMillis(): Long

class CrossPlatformEventBus {
    private val _events = Channel<String>(Channel.UNLIMITED)
    val events: ReceiveChannel<String> = _events
    
    suspend fun publish(event: String) {
        _events.send(event)
    }
    
    fun close() {
        _events.close()
    }
}

// platformMain (JVM)
actual fun currentTimeMillis(): Long = System.currentTimeMillis()

// platformMain (iOS/Native)
actual fun currentTimeMillis(): Long = kotlinx.datetime.Clock.System.now().toEpochMilliseconds()

// 使用示例
class CrossPlatformApp {
    private val eventBus = CrossPlatformEventBus()
    
    fun start(scope: CoroutineScope) {
        scope.launch {
            eventBus.events.consumeEach { event ->
                println("[$currentTimeMillis()] Event: $event")
            }
        }
    }
    
    suspend fun emit(event: String) {
        eventBus.publish(event)
    }
}

5. 对比分析#

5.1 Kotlin Channel vs Java BlockingQueue#

5.2 Kotlin Channel vs Go channel#

5.3 Kotlin Channel vs Rust mpsc#

5.4 Kotlin Channel vs Scala/Akka Actor#

5.5 BroadcastChannel vs SharedFlow 详细对比#

5.6 Channel vs Flow vs SharedFlow vs StateFlow#

5.7 性能基准对比#

在 8 核 CPU、16GB RAM 的测试环境中，使用 JMH 基准测试：

观察：

• Kotlin Channel 性能略低于 Go/Java/Rust，主要因为协程挂起开销。
• 内存占用较高，因为 Continuation 对象分配。
• 但 Channel 在”挂起不阻塞线程”方面有独特优势，适合高并发协程场景。

6. 常见陷阱与最佳实践#

6.1 陷阱 1：忘记关闭 Channel 导致泄漏#

// 反模式：忘记关闭 Channel
fun badExample() = CoroutineScope(Dispatchers.Default).launch {
    val channel = Channel<Int>()
    
    launch {
        repeat(10) { channel.send(it) }
        // 忘记调用 channel.close()
    }
    
    launch {
        channel.consumeEach { println(it) }
        // 消费者会一直等待，因为 Channel 没有关闭
    }
}

// 正确做法：使用 produce 或显式关闭
fun goodExample() = CoroutineScope(Dispatchers.Default).launch {
    val channel = produce {
        repeat(10) { send(it) }
        // produce 自动关闭 Channel
    }
    
    channel.consumeEach { println(it) }
}

最佳实践：优先使用 produce { ... } 构建器，它会自动管理 Channel 生命周期。

6.2 陷阱 2：在 GlobalScope 中使用 Channel#

// 反模式：GlobalScope 中创建 Channel
val globalChannel = GlobalScope.produce {
    send(1)
}
// 全局协程无法被取消，Channel 可能泄漏

// 正确做法：使用结构化并发
suspend fun goodExample(scope: CoroutineScope) {
    val channel = scope.produce {
        send(1)
    }
    channel.consumeEach { println(it) }
}

最佳实践：始终在 CoroutineScope 内创建 Channel，避免 GlobalScope。

6.3 陷阱 3：误解 RENDEZVOUS 模式#

// 反模式：RENDEZVOUS 模式下接收方未启动
suspend fun badExample() = coroutineScope {
    val channel = Channel<Int>()  // RENDEZVOUS
    
    launch {
        channel.send(1)  // 永远挂起，因为没有接收方
        println("Sent")  // 永远不会执行
    }
    
    delay(1000)
    println("Done")  // 输出 Done，但发送方仍挂起
}

// 正确做法：先启动接收方，或使用 BUFFERED
suspend fun goodExample() = coroutineScope {
    val channel = Channel<Int>(Channel.BUFFERED)  // 有缓冲
    
    launch {
        channel.send(1)  // 立即返回
        println("Sent")
    }
    
    delay(100)
    val v = channel.receive()
    println("Received $v")
}

最佳实践：理解 RENDEZVOUS 的会合语义，必要时使用 BUFFERED 解耦。

6.4 陷阱 4：在 select 中使用阻塞操作#

// 反模式：select 中调用阻塞操作
select<Unit> {
    channel1.onReceive { 
        Thread.sleep(1000)  // 阻塞！会阻塞整个 select
    }
    channel2.onReceive { ... }
}

// 正确做法：使用 suspend 函数
select<Unit> {
    channel1.onReceive { 
        delay(1000)  // 挂起，不阻塞线程
    }
    channel2.onReceive { ... }
}

最佳实践：select 的 clause 中只调用 suspend 函数，避免阻塞。

6.5 陷阱 5：使用废弃的 BroadcastChannel#

// 反模式：使用废弃的 BroadcastChannel
val broadcast = BroadcastChannel<Int>(10)  // 废弃！

// 正确做法：使用 SharedFlow
val sharedFlow = MutableSharedFlow<Int>(
    replay = 0,
    extraBufferCapacity = 10,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)

最佳实践：迁移到 SharedFlow，享受统一的 Flow API。

6.6 陷阱 6：使用废弃的 actor 构建器#

// 反模式：使用废弃的 actor 构建器
val counter = actor<Int> {
    var count = 0
    for (msg in channel) {
        count += msg
    }
}

// 正确做法：使用 Channel + 状态封装协程
class CounterActor {
    private val channel = Channel<Int>()
    private var count = 0
    
    fun start(scope: CoroutineScope) {
        scope.launch {
            for (msg in channel) {
                count += msg
            }
        }
    }
    
    suspend fun increment(by: Int = 1) {
        channel.send(by)
    }
}

最佳实践：手动封装 Channel 与协程，更灵活可控。

6.7 陷阱 7：在 Channel 上调用多次 receive#

// 反模式：在 for 循环外手动 receive
val channel = Channel<Int>()
launch {
    repeat(10) { channel.send(it) }
    channel.close()
}

// 错误：手动 receive 可能错过异常处理
while (true) {
    val v = channel.receive()  // Channel 关闭后会抛异常
    println(v)
}

// 正确做法：使用 for 循环或 consumeEach
for (v in channel) {
    println(v)
}
// 或
channel.consumeEach { println(it) }

最佳实践：使用 for (v in channel) 或 consumeEach，自动处理关闭。

6.8 陷阱 8：异常吞没#

// 反模式：异常未被处理
val channel = Channel<Int>()
launch {
    try {
        channel.send(1)
    } catch (e: Exception) {
        // 吞掉异常
    }
}

// 正确做法：使用 CoroutineExceptionHandler 或传播异常
val handler = CoroutineExceptionHandler { _, e ->
    println("Caught: $e")
}

launch(handler) {
    channel.send(1)
}

最佳实践：使用 CoroutineExceptionHandler 或 SupervisorJob 处理异常。

6.9 陷阱 9：混淆 Channel 与 SharedFlow#

// 反模式：用 Channel 做事件广播
val eventChannel = Channel<Event>()
// 多个订阅者无法同时接收，因为 Channel 是点对点的

// 正确做法：用 SharedFlow 做广播
val eventFlow = MutableSharedFlow<Event>()

最佳实践：

• 点对点消息：Channel。
• 多播事件：SharedFlow。
• 状态表示：StateFlow。
• 数据流转换：Flow。

6.10 陷阱 10：在 consumeEach 外消费 Channel#

// 反模式：consumeEach 与手动 receive 混用
val channel = Channel<Int>()

launch {
    channel.consumeEach { 
        println(it) 
    }
}

launch {
    val v = channel.receive()  // 可能抛出 IllegalStateException
}

最佳实践：consumeEach 会取消 Channel，避免在 consumeEach 外消费。

6.11 最佳实践总结#

1. 优先使用 produce：自动管理 Channel 生命周期。
2. 结构化并发：始终在 CoroutineScope 内创建 Channel。
3. 显式关闭：手动创建的 Channel 必须显式 close()。
4. 正确选择容量：默认 RENDEZVOUS，需要解耦时用 BUFFERED，需要最新值时用 CONFLATED。
5. 避免 GlobalScope：使用结构化并发管理生命周期。
6. 使用 SharedFlow 替代 BroadcastChannel：现代 API 更统一。
7. 避免在 select 中阻塞：只调用 suspend 函数。
8. 错误处理：使用 receiveCatching 或 CoroutineExceptionHandler。
9. 性能优化：高吞吐量场景考虑 UNLIMITED，但要监控内存。
10. 类型安全：使用 sealed class 表达消息类型，编译期检查。

7. 工程实践#

7.1 Gradle 项目配置#

// build.gradle.kts
plugins {
    kotlin("jvm") version "2.0.0"
    kotlin("plugin.serialization") version "2.0.0"
    application
}

dependencies {
    // 协程核心库
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.0")
    
    // 测试支持
    testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.8.0")
    testImplementation("org.junit.jupiter:junit-jupiter:5.10.0")
    testImplementation("org.jetbrains.kotlin:kotlin-test:2.0.0")
}

application {
    mainClass.set("com.example.ApplicationKt")
}

tasks.test {
    useJUnitPlatform()
}

7.2 Android 项目配置#

// android/app/build.gradle.kts
plugins {
    id("com.android.application")
    id("org.jetbrains.kotlin.android") version "2.0.0"
}

dependencies {
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.0")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-android:1.8.0")
    
    // Lifecycle scope 支持
    implementation("androidx.lifecycle:lifecycle-viewmodel-ktx:2.8.0")
    implementation("androidx.lifecycle:lifecycle-runtime-ktx:2.8.0")
}

7.3 KMP 项目配置#

// build.gradle.kts
plugins {
    kotlin("multiplatform") version "2.0.0"
}

kotlin {
    jvm()
    js { browser() }
    iosX64()
    iosArm64()
    iosSimulatorArm64()
    
    sourceSets {
        commonMain.dependencies {
            implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.0")
        }
        
        commonTest.dependencies {
            implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.8.0")
        }
    }
}

7.4 单元测试示例#

// src/test/kotlin/ChannelTest.kt
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.test.*
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.Assertions.*
import kotlin.coroutines.ContinuationInterceptor

class ChannelTest {
    
    @Test
    fun `should send and receive values in order`() = runTest {
        val channel = Channel<Int>(Channel.UNLIMITED)
        
        // 发送方
        launch {
            channel.send(1)
            channel.send(2)
            channel.send(3)
            channel.close()
        }
        
        // 接收并验证顺序
        val results = mutableListOf<Int>()
        channel.consumeEach { results.add(it) }
        
        assertEquals(listOf(1, 2, 3), results)
    }
    
    @Test
    fun `should block send when buffer is full`() = runTest {
        val channel = Channel<Int>(capacity = 2)
        
        // 先填满缓冲区
        channel.trySend(1).assertSuccess()
        channel.trySend(2).assertSuccess()
        
        // 第三次应该失败（缓冲满）
        val result = channel.trySend(3)
        assertTrue(result.isFailure)
        
        // 接收后再发送
        assertEquals(1, channel.receive())
        channel.trySend(3).assertSuccess()
        assertEquals(2, channel.receive())
        assertEquals(3, channel.receive())
    }
    
    @Test
    fun `should receive latest value in CONFLATED mode`() = runTest {
        val channel = Channel<Int>(Channel.CONFLATED)
        
        channel.trySend(1)
        channel.trySend(2)
        channel.trySend(3)
        
        assertEquals(3, channel.receive())
    }
    
    @Test
    fun `should close channel and throw on receive`() = runTest {
        val channel = Channel<Int>()
        channel.close()
        
        assertTrue(channel.isClosedForReceive)
        
        val result = channel.receiveCatching()
        assertTrue(result.isClosed)
    }
    
    @Test
    fun `should select first ready channel`() = runTest {
        val channel1 = Channel<String>()
        val channel2 = Channel<String>()
        
        channel2.send("from-2")  // channel2 先就绪
        
        val result = select<String> {
            channel1.onReceive { it }
            channel2.onReceive { it }
        }
        
        assertEquals("from-2", result)
    }
}

private fun ChannelResult<Unit>.assertSuccess() {
    assertTrue(isSuccess, "Expected success but got: ${exceptionOrNull()}")
}

7.5 性能监控#

// 性能监控工具
class ChannelMonitor<T>(private val channel: Channel<T>) {
    private var sentCount = 0L
    private var receivedCount = 0L
    private var lastSentAt: Long = 0
    private var lastReceivedAt: Long = 0
    
    suspend fun sendWithMonitor(value: T) {
        val start = System.nanoTime()
        channel.send(value)
        val end = System.nanoTime()
        sentCount++
        lastSentAt = end
        if (sentCount % 1000 == 0L) {
            println("[Monitor] Sent: $sentCount, latency: ${(end - start) / 1000} μs")
        }
    }
    
    suspend fun receiveWithMonitor(): T {
        val start = System.nanoTime()
        val value = channel.receive()
        val end = System.nanoTime()
        receivedCount++
        lastReceivedAt = end
        if (receivedCount % 1000 == 0L) {
            println("[Monitor] Received: $receivedCount, latency: ${(end - start) / 1000} μs")
        }
        return value
    }
    
    fun getStats(): ChannelStats {
        return ChannelStats(
            sentCount = sentCount,
            receivedCount = receivedCount,
            pendingInBuffer = sentCount - receivedCount,
            lastSentAt = lastSentAt,
            lastReceivedAt = lastReceivedAt
        )
    }
}

data class ChannelStats(
    val sentCount: Long,
    val receivedCount: Long,
    val pendingInBuffer: Long,
    val lastSentAt: Long,
    val lastReceivedAt: Long
)

7.6 集成测试示例#

// 集成测试：完整的生产者-消费者系统
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.Assertions.*
import java.util.concurrent.atomic.AtomicInteger

class ProducerConsumerIntegrationTest {
    
    @Test
    fun `should process orders end-to-end`() = runBlocking {
        val orderCount = AtomicInteger(0)
        val processedCount = AtomicInteger(0)
        
        val orderChannel = Channel<String>(capacity = 10)
        val resultChannel = Channel<String>(capacity = 10)
        
        // 生产者
        val producer = launch {
            repeat(100) { i ->
                val order = "ORDER-$i"
                orderChannel.send(order)
                orderCount.incrementAndGet()
            }
            orderChannel.close()
        }
        
        // 多个消费者
        val consumers = (1..4).map { workerId ->
            launch {
                for (order in orderChannel) {
                    delay(10)  // 模拟处理
                    resultChannel.send("$order-processed-by-$workerId")
                    processedCount.incrementAndGet()
                }
            }
        }
        
        // 结果收集器
        val collector = launch {
            for (result in resultChannel) {
                // 验证结果格式
                assertTrue(result.matches(Regex("ORDER-\\d+-processed-by-\\d+")))
            }
        }
        
        // 等待所有完成
        producer.join()
        consumers.forEach { it.join() }
        resultChannel.close()
        collector.join()
        
        assertEquals(100, orderCount.get())
        assertEquals(100, processedCount.get())
    }
    
    @Test
    fun `should handle backpressure with BUFFERED channel`() = runBlocking {
        val channel = Channel<Int>(capacity = 10)
        val sentTimes = mutableListOf<Long>()
        val receivedTimes = mutableListOf<Long>()
        
        // 快速生产者
        val producer = launch {
            repeat(50) { i ->
                channel.send(i)
                sentTimes.add(System.currentTimeMillis())
            }
            channel.close()
        }
        
        // 慢速消费者（背压）
        val consumer = launch {
            for (v in channel) {
                delay(50)  // 每个元素处理 50ms
                receivedTimes.add(System.currentTimeMillis())
            }
        }
        
        producer.join()
        consumer.join()
        
        // 验证背压：发送方应在缓冲满后等待
        assertEquals(50, sentTimes.size)
        assertEquals(50, receivedTimes.size)
    }
}

7.7 调试技巧#

// 启用协程调试
suspend fun main() = runBlocking {
    // 设置调试代理
    System.setProperty("kotlinx.coroutines.debug", "on")
    
    val channel = Channel<String>()
    
    // 使用 CoroutineName 标识协程
    launch(CoroutineName("producer-1")) {
        repeat(5) { i ->
            val msg = "msg-$i from ${coroutineContext[CoroutineName]}"
            channel.send(msg)
            delay(100)
        }
        channel.close()
    }
    
    launch(CoroutineName("consumer-1")) {
        channel.consumeEach { msg ->
            println("[${Thread.currentThread().name}] $msg")
        }
    }
    
    delay(1000)
}

输出示例：

[main @producer-1#2] msg-0 from CoroutineName(producer-1)
[main @consumer-1#3] msg-0 from CoroutineName(producer-1)
...

7.8 生产部署注意事项#

1. 资源监控：监控 Channel 的缓冲区大小，避免 UNLIMITED 导致 OOM。
2. 错误恢复：使用 SupervisorJob 隔离故障，避免单点故障传播。
3. 优雅关闭：在 Runtime.addShutdownHook 中关闭所有 Channel。
4. 背压策略：根据业务场景选择 SUSPEND/DROP_OLDEST/DROP_LATEST。
5. 指标收集：暴露 Channel 的指标（sent/received/pending）到 Prometheus。
6. 日志隔离：为每个 Channel 操作添加 traceId，便于追踪。
7. 测试覆盖：单元测试覆盖所有容量模式、关闭场景、异常场景。

8. 案例研究#