当前位置：首页 > news >正文

Channel 和 Flow 选择场景对比（例子：不停发事件的场景）

news 2025/10/8 6:46:07

引言

对于不停发事件的场景，选择 Channel 还是 Flow 取决于具体需求。让我用一个详细的对比表格来决策：

特性	Channel	Flow
数据性质	热数据流	冷数据流
消费者关系	竞争消费（一个消息只被一个消费者处理）	独立消费（每个消费者获得完整数据流）
背压处理	有界队列，可阻塞或丢弃	天然支持背压，按需生产
内存占用	可能堆积消息在内存中	按需生产，内存友好
适用场景	事件总线、任务队列、实时消息	数据转换、状态推导、响应式流

具体选择指南

选择 Channel 的情况 ✅

当你的场景是"事件"而不是"状态"时：

// 事件总线模式 - 适合用 Channel
object EventBus {private val eventChannel = Channel<Event>(Channel.BUFFERED)suspend fun sendEvent(event: Event) {eventChannel.send(event)}fun receiveEvents(): Channel<Event> = eventChannel
}// 使用场景：多个消费者竞争处理事件
fun main() = runBlocking {// 多个消费者repeat(3) { index ->launch {for (event in EventBus.receiveEvents()) {println("消费者$index 处理: $event")// 每个事件只被一个消费者处理}}}// 生产者不停发送事件repeat(10) { i ->EventBus.sendEvent(Event("事件$i"))delay(100)}
}

任务队列场景：

class TaskProcessor {private val taskChannel = Channel<Task>(capacity = 100)init {// 启动固定数量的工作协程repeat(4) { id ->launch {for (task in taskChannel) {processTask(task, id)}}}}suspend fun submitTask(task: Task) {taskChannel.send(task)}private fun processTask(task: Task, workerId: Int) {println("Worker $workerId 处理: $task")}
}

选择 Flow 的情况 ✅

当需要每个消费者获得完整数据流时：

// 传感器数据流 - 适合用 Flow
class SensorManager {fun sensorDataFlow(): Flow<SensorData> = flow {while (true) {val data = readSensorData()emit(data)delay(10) // 10ms 采样间隔}}
}// 使用场景：多个消费者都需要完整数据
fun main() = runBlocking {val sensorFlow = SensorManager().sensorDataFlow()// 消费者1：显示数据launch {sensorFlow.collect { data ->updateDisplay(data)}}// 消费者2：记录日志launch {sensorFlow.collect { data ->writeToLog(data)}}// 消费者3：分析数据launch {sensorFlow.collect { data ->analyzeData(data)}}
}

数据转换和处理场景：

fun dataProcessingFlow(): Flow<ProcessedData> = flow {// 模拟不停产生原始数据var count = 0while (true) {emit(RawData(count++))delay(50)}
}
.buffer() // 处理背压
.map { rawData ->// 转换数据ProcessedData(rawData.value * 2)
}
.filter { it.value % 3 == 0 } // 过滤数据
.distinctUntilChanged() // 去重

混合方案：SharedFlow

对于"不停发事件"的场景，SharedFlow 通常是最佳选择，它结合了两者的优点：

class EventManager {// SharedFlow - 专为事件流设计private val _events = MutableSharedFlow<Event>(replay = 0, // 新订阅者不接收历史事件extraBufferCapacity = 64, // 缓冲区大小onBufferOverflow = BufferOverflow.DROP_OLDEST // 背压策略)val events: SharedFlow<Event> = _events.asSharedFlow()suspend fun emitEvent(event: Event) {_events.emit(event)}// 或者非挂起版本fun tryEmitEvent(event: Event) {_events.tryEmit(event)}
}// 使用：多个订阅者都能收到事件
fun main() = runBlocking {val eventManager = EventManager()// 订阅者1launch {eventManager.events.collect { event ->println("订阅者1收到: $event")}}// 订阅者2launch {eventManager.events.collect { event ->println("订阅者2收到: $event")}}// 不停发送事件var eventCount = 0while (true) {eventManager.emitEvent(Event("事件${eventCount++}"))delay(100)}
}

决策流程图

不停发事件的场景
↓
是事件(多个消费者竞争)还是状态(每个消费者都要)?
↓                                 ↓
竞争消费                          独立消费
↓                                 ↓
选择 Channel                  选择 Flow/SharedFlow
↓ ↓
任务队列、消息总线          数据流、状态更新、传感器数据