Android中BroadcastChannel.asFlow()无法使用，不用StateFlow/SharedFlow的解决办法

如果不想用官方推荐的StateFlow/SharedFlow，下面是两种可行的替代方案：

方案1：用callbackFlow手动实现广播逻辑 #

通过callbackFlow自定义支持多订阅者的广播流，每个新订阅者都会收到后续发送的所有事件：

import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.launch
import java.util.concurrent.ConcurrentLinkedQueue

// 定义广播事件类型
data class BroadcastEvent(val content: String)

// 线程安全的订阅者集合，避免多线程操作问题
private val subscribers = ConcurrentLinkedQueue<(BroadcastEvent) -> Unit>()

// 构建广播Flow
val broadcastFlow: Flow<BroadcastEvent> = callbackFlow {
    // 当前订阅者的事件回调
    val subscriber: (BroadcastEvent) -> Unit = { event ->
        launch { send(event) }
    }
    // 注册订阅者
    subscribers.add(subscriber)
    // Flow取消时自动注销订阅者
    awaitClose { subscribers.remove(subscriber) }
}

// 对外暴露的事件发送方法
fun sendBroadcastEvent(event: BroadcastEvent) {
    subscribers.forEach { it(event) }
}

使用示例 #

import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.launch

// 第一个订阅者
GlobalScope.launch {
    broadcastFlow.collect {
        println("订阅者1收到：${it.content}")
    }
}

// 第二个订阅者
GlobalScope.launch {
    broadcastFlow.collect {
        println("订阅者2收到：${it.content}")
    }
}

// 发送测试事件
sendBroadcastEvent(BroadcastEvent("这是一条广播消息"))

方案2：用普通Channel + 多协程转发实现广播 #

如果依赖普通Channel，可通过为每个订阅者启动独立协程转发消息的方式实现广播：

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

private val sourceChannel = Channel<BroadcastEvent>(Channel.UNLIMITED)

val broadcastFlow: Flow<BroadcastEvent> = flow {
    // 每个订阅者都启动协程从源Channel取消息并发送到Flow
    val job = launch {
        for (event in sourceChannel) {
            emit(event)
        }
    }
    // Flow取消时停止转发协程
    awaitClose { job.cancel() }
}

// 发送事件方法
fun sendEvent(event: BroadcastEvent) {
    sourceChannel.trySend(event)
}

注意点 #

• 方案1的订阅者集合用ConcurrentLinkedQueue保证线程安全，避免多线程添加/删除时的并发问题
• 方案2中每个订阅者会独立消费源Channel的消息，因此源Channel需要用UNLIMITED容量，避免消息丢失
• 如果项目已引入RxJava，也可以用PublishSubject实现广播，再通过asFlow()转换为Flow使用，适合已有RxJava技术栈的项目

内容的提问来源于stack exchange，提问作者kunal