12、Channel的高级用法：BroadcastChannel（已废弃）与SharedFlow替代、Channel的fan-out与fan-in模式、Channel的背压策略

各位同学，今天我们来聊聊Channel的高级玩法。说实话，Channel在协程世界里就像水管工手里的接头——看似简单，但接错了地方，整个系统就崩了。我当年刚接触协程时，就在BroadcastChannel上栽过跟头，后来才发现官方早就给了更好的方案。

一、BroadcastChannel已废弃，为什么？

先说说这个“过气网红”。BroadcastChannel，顾名思义，就是一对多的广播通道。一个生产者发消息，多个消费者都能收到。听起来很美好对吧？

但问题出在哪呢？

• 线程安全问题：BroadcastChannel内部实现有竞态条件，多线程环境下容易出bug
• 背压处理不完善：消费者消费速度不一致时，要么丢数据，要么内存暴涨
• 生命周期耦合：Channel关闭后，所有订阅者都得重新订阅，很麻烦

二、SharedFlow：官方推荐的替代方案

SharedFlow是Flow家族的一员，专门解决一对多场景。它比BroadcastChannel强在哪？我列几个关键点：

说白了，SharedFlow就是BroadcastChannel的“完全体”。你想想看，一个支持重放、支持背压、线程安全的事件总线，谁不爱？

SharedFlow的基本用法

// 创建一个SharedFlow，replay=1表示新订阅者能收到最近一条消息
val sharedFlow = MutableSharedFlow<String>(
    replay = 1,
    extraBufferCapacity = 10,
    onBufferOverflow = BufferOverflow.SUSPEND
)

// 生产者
fun produce() {
    CoroutineScope(Dispatchers.Default).launch {
        repeat(100) {
            sharedFlow.emit("消息$it")
            delay(100)
        }
    }
}

// 消费者1
fun consume1() {
    CoroutineScope(Dispatchers.IO).launch {
        sharedFlow.collect { msg ->
            println("消费者1收到：$msg")
        }
    }
}

// 消费者2
fun consume2() {
    CoroutineScope(Dispatchers.IO).launch {
        sharedFlow.collect { msg ->
            println("消费者2收到：$msg")
        }
    }
}

三、Channel的fan-out与fan-in模式

这两个模式，说白了就是“分”和“合”的艺术。我在做实时数据管道时经常用到。

Fan-out：一对多分发

一个生产者，多个消费者。每个消费者都能收到全部消息。SharedFlow天然支持fan-out。

// Fan-out示例：一个生产者，两个消费者
val channel = Channel<String>(Channel.UNLIMITED)

fun producer() {
    CoroutineScope(Dispatchers.Default).launch {
        repeat(10) {
            channel.send("数据$it")
            delay(50)
        }
        channel.close()
    }
}

fun consumer(name: String) {
    CoroutineScope(Dispatchers.IO).launch {
        for (msg in channel) {
            println("$name 处理：$msg")
            delay(100) // 模拟处理耗时
        }
    }
}

// 启动两个消费者
consumer("消费者A")
consumer("消费者B")
producer()

注意：这里的Channel是普通的Channel，不是BroadcastChannel。每个消息只能被一个消费者消费。如果想广播，还是用SharedFlow。

Fan-in：多对一汇聚

多个生产者，一个消费者。消息汇聚到一个通道里。这个模式在合并多个数据源时特别有用。

// Fan-in示例：两个生产者，一个消费者
val resultChannel = Channel<String>(Channel.BUFFERED)

fun producer1() {
    CoroutineScope(Dispatchers.Default).launch {
        repeat(5) {
            resultChannel.send("生产者1-数据$it")
            delay(80)
        }
    }
}

fun producer2() {
    CoroutineScope(Dispatchers.Default).launch {
        repeat(5) {
            resultChannel.send("生产者2-数据$it")
            delay(120)
        }
    }
}

fun consumer() {
    CoroutineScope(Dispatchers.IO).launch {
        for (msg in resultChannel) {
            println("消费者收到：$msg")
        }
    }
}

producer1()
producer2()
consumer()

嗯，这里要注意：多个生产者往同一个Channel发消息时，Channel内部会做线程安全处理，不会出现数据错乱。但如果你用Channel.UNLIMITED，小心内存溢出。

四、Channel的背压策略

背压，说白了就是“生产者太快，消费者太慢”的问题。你想想看，如果生产者每秒发1000条消息，消费者每秒只能处理10条，那中间积压的数据怎么办？

Channel提供了几种背压策略：

// 背压策略示例
val channel = Channel<String>(
    capacity = 10,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)

fun fastProducer() {
    CoroutineScope(Dispatchers.Default).launch {
        repeat(100) {
            channel.trySend("消息$it")
            println("发送：消息$it")
            delay(10)
        }
    }
}

fun slowConsumer() {
    CoroutineScope(Dispatchers.IO).launch {
        for (msg in channel) {
            println("消费：$msg")
            delay(200) // 很慢的消费者
        }
    }
}

fastProducer()
slowConsumer()

五、知识体系总览

下面这张图，是我梳理的Channel高级用法知识体系。你可以把它当作思维导图来用。

这张图把今天讲的内容串起来了。从BroadcastChannel的废弃，到SharedFlow的替代，再到fan-out/fan-in模式，最后是背压策略。你想想看，这些知识点其实都围绕一个核心问题：如何高效、安全地在协程间传递数据。

好了，Channel的高级用法就聊到这里。记住一句话：选对工具，事半功倍。SharedFlow是广播场景的首选，背压策略要根据业务场景灵活选择，fan-out和fan-in是数据管道的基石。下次遇到类似需求，你应该知道怎么选了。