从不同Flow收集数据时是否可在协程中调用emit方法？

核心结论 #

不要在onEach中单独启动协程调用emit，这种写法会引入事件乱序、协程泄漏的问题，也完全没有必要。

具体问题解释 #

关于emit的线程安全 #

官方文档提到的emit非线程安全，指的是不能在非协程的裸线程环境下无同步调用emit。只要你是在协程上下文内调用MutableSharedFlow的emit方法，它本身就是线程安全的，不需要额外启动协程做隔离。
你写的launch { flowToReturn.emit(it) }写法反而会导致事件发送顺序不可控，不同协程的调度顺序完全由调度器决定，最终flowToReturn收到的事件和原连接流的顺序不一致。同时如果emit挂起，这些零散启动的子协程会一直存活，连接断开重连后也不会自动取消，会造成严重的协程泄漏。

关于无订阅者时emit挂起的问题 #

你担心emit挂起影响下游doStuff逻辑执行的解决方案，根本不是启动新协程，而是通过配置MutableSharedFlow的参数避免emit挂起：

// 配置1：允许丢事件的场景，永远不会挂起
val flowToReturn = MutableSharedFlow<Event>(
    extraBufferCapacity = 64, // 根据事件吞吐量调整缓存大小
    onBufferOverflow = BufferOverflow.DROP_OLDEST // 缓存满时丢弃最旧的事件，也可根据需求选DROP_LATEST
)

// 配置2：不允许丢事件的场景，只要内存足够就不会挂起
val flowToReturn = MutableSharedFlow<Event>(
    extraBufferCapacity = Int.MAX_VALUE,
    onBufferOverflow = BufferOverflow.SUSPEND
)

如果你的业务允许少量事件丢失，甚至可以直接调用非挂起的tryEmit方法，不用改SharedFlow配置：

.onEach { flowToReturn.tryEmit(it) }

推荐实现方案 #

更合理的写法是将事件转发和业务逻辑处理拆分为两个独立的收集流程，彻底避免两个逻辑的耦合，同时保证重连逻辑正常运行：

scope.launch {
    val flowToReturn = MutableSharedFlow<Event>(
        extraBufferCapacity = 256,
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    )
    while (true) {
        val sharedConnectionFlow = client
            .connect() // returns Flow<Event>
            .catch { /* 错误处理逻辑 */ }
            // 将连接流转为共享流，多个收集器共用同一个连接
            .shareIn(this, SharingStarted.WhileSubscribed())

        // 启动独立协程处理事件转发，父协程是当前循环协程，重连时自动取消
        launch {
            sharedConnectionFlow.collect {
                flowToReturn.emit(it)
            }
        }

        // 主业务逻辑收集，不受转发逻辑影响
        sharedConnectionFlow
            .filterIsInstance<Event.Some>()
            .collect { someEvent ->
                doStuff(someEvent)
            }
    }
}.start()

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