如何在Kotlin协程中创建类似RxJava PublishSubject的MutableSharedFlow？

Kotlin标准协程库没有提供和PublishSubject命名完全对应的内置类，但可以通过自定义配置MutableSharedFlow实现完全一致的行为，无需额外引入第三方组件。

你提到的默认MutableSharedFlow发射时需要等待收集器处理的问题，是因为其默认配置使用了BufferOverflow.SUSPEND的背压策略，我们可以通过调整参数对齐PublishSubject的所有核心特性：

• 支持多收集器同时订阅，所有收集器都能收到订阅后发射的所有值
• 发射值不需要等待收集器处理完前序值，不会阻塞发射线程
• 新订阅者不会收到订阅之前发射的历史值

核心实现代码 #

import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.BufferOverflow

fun <T> PublishFlow(): MutableSharedFlow<T> = MutableSharedFlow(
    replay = 0,
    extraBufferCapacity = Int.MAX_VALUE,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)

参数说明 #

• replay = 0：保证新加入的收集器仅能收到订阅完成后发射的新值，和PublishSubject新订阅者无法接收历史事件的行为完全对齐。
• extraBufferCapacity = Int.MAX_VALUE：提供足够大的缓冲空间，常规使用场景下不会出现缓冲区不足的问题，发射时不会触发挂起。
• onBufferOverflow = BufferOverflow.DROP_OLDEST：极端场景下缓冲区占满时，直接丢弃最旧的未处理值，不会阻塞发射逻辑，完全匹配PublishSubject无背压、发射不等待消费者的特性。

使用说明 #

如果你不希望下游拿到可发射的MutableSharedFlow实例，可以将其向上转型为普通SharedFlow暴露，和RxJava中把PublishSubject转型为Observable给下游使用的逻辑一致：

private val _eventFlow = PublishFlow<String>()
val eventFlow: SharedFlow<String> = _eventFlow

// 发射端使用
suspend fun sendEvent(event: String) {
    _eventFlow.emit(event)
}

如果你需要对齐PublishSubject的onComplete、onError终止事件能力，可以用密封类包装事件类型：

sealed class FlowEvent<out T> {
    data class Next<T>(val value: T) : FlowEvent<T>()
    data class Error(val throwable: Throwable) : FlowEvent<Nothing>()
    object Complete : FlowEvent<Nothing>()
}

// 声明对应类型的PublishFlow
private val _eventFlow = PublishFlow<FlowEvent<String>>()

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