Android Kotlin Channel 与并发通信模式

Channel 基础

Channel 类型	特点
`Channel.UNLIMITED`	无界，容易 OOM
`Channel.BUFFERED`	默认缓冲 64，常用
`Channel.CONFLATED`	只保留最新值
`Channel.RENDEZVOUS`	无缓冲，发送/接收需要配对

val channel = Channel<Int>(capacity = Channel.BUFFERED)

launch {
    repeat(10) { channel.send(it) }
    channel.close()
}

launch {
    for (value in channel) {
        println(value)
    }
}

生产者/消费者模式

fun CoroutineScope.produceEvents(): ReceiveChannel<Event> = produce(capacity = Channel.BUFFERED) {
    while (isActive) {
        send(Event(System.currentTimeMillis()))
        delay(1000)
    }
}

val events = produceEvents()
launch {
    events.consumeEach { handleEvent(it) }
}

背压控制

限制发送速率

val channel = Channel<Task>(capacity = 10)

launch {
    channel.consumeEach { task ->
        semaphore.withPermit { process(task) }
    }
}

使用 `Flow` 转换

val flow = channel.receiveAsFlow()
    .buffer(Channel.CONFLATED)
    .collectLatest { process(it) }

Select 表达式

select<Unit> {
    channel.onReceiveCatching { value ->
        value.getOrNull()?.let(::handle)
    }
    timeoutChannel.onReceive { handleTimeout() }
}

Android 场景

1. 上传队列

class UploadManager {
    private val channel = Channel<UploadTask>(Channel.BUFFERED)
    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)

    init {
        repeat(3) {
            scope.launch { workerLoop() }
        }
    }

    fun enqueue(task: UploadTask) {
        scope.launch { channel.send(task) }
    }

    private suspend fun workerLoop() {
        for (task in channel) {
            runCatching { uploader.upload(task) }
                .onFailure { delay(3000); channel.send(task) } // 重试
        }
    }
}

2. UI 事件节流

val clickChannel = Channel<Unit>(Channel.CONFLATED)

LaunchedEffect(Unit) {
    clickChannel.consumeAsFlow().debounce(300).collect { onConfirm() }
}

Button(onClick = { clickChannel.trySend(Unit) }) { Text("确认") }

调试与测试

@Test
fun `channel processes tasks sequentially`() = runTest {
    val results = mutableListOf<Int>()
    val channel = Channel<Int>(Channel.RENDEZVOUS)

    val job = launch {
        for (value in channel) {
            delay(10)
            results += value
        }
    }

    channel.send(1)
    channel.send(2)
    channel.close()
    job.join()
    assertEquals(listOf(1, 2), results)
}

常见问题

问题	原因	解决方案
Channel 泄漏	未关闭或未消费	在适当时机调用 `close()`，确保消费者退出
OOM	无界缓冲导致堆积	使用 `capacity` 限制或 `Flow` 背压
死锁	同步发送/接收阻塞	使用 `launch` 嵌套或 `Channel.BUFFERED`

实战案例：什么时候该用 Channel 而不是 Flow

官方文档里有句很关键的话：Channel 是”热”的、一点对点的；Flow 是”冷”的、一对多的。但这句话太抽象，我换个说法：

Channel = 对讲机：一端说，另一端必须有人听。没人听，说话的人就卡住（Rendezvous）。适合”任务分发”——比如把下载任务丢给后台 worker，一个任务只被一个人领走。
Flow = 广播电台：没人听也照播，有人调频（collect）才从头开始播。适合”状态观察”——比如 UI 观察 ViewModel 的 UiState，晚进来的观察者也能拿到最新值（StateFlow）。

我在 Flutter 里踩过对等的坑：用 broadcast Stream 做”一次性事件”（比如”下载完成”通知），结果页面重建后 listener 重新订阅，旧事件又触发一次——这就是著名的”事件重放”问题。Kotlin 社区为此专门有个讨论：要么用 Channel（事件只消费一次），要么用 SharedFlow 配 replay=0。Google 官方的 Now in Android 项目里，UiEvent 就是用 Channel 发的。

所以选型口诀：状态用 StateFlow，事件用 Channel。把事件放进 StateFlow，是 Android 开发里最高频的 bug 来源之一。

参考：Kotlin 官方文档 Channels vs Flow 说明；Now in Android 开源项目

总结

Channel 适合构建生产者-消费者模型、实现异步队列。结合 Flow、Select 可构建复杂并发模式。合理设置容量、关闭通道并编写单元测试，能保证 Channel 在 Android 项目中的稳定性。