从不同的流程收集时，我应该从协程中发出吗？答案

【问题标题】：Should I emit from a coroutine when collecting from a different flow?从不同的流程收集时，我应该从协程中发出吗？
【发布时间】：2021-11-27 14:49:16
【问题描述】：

我有一个用例，我需要触发从流中收集的特定事件并在它关闭时重新启动它。我还需要将所有事件发送到不同的流程。我当前的实现如下所示：

scope.launch {
    val flowToReturn = MutableSharedFlow<Event>()
    while (true) {
        client
            .connect()                                   // returns Flow<Event>
            .catch { ... }                               // ignore errors
            .onEach { launch { flowToReturn.emit(it) } } // problem here
            .filterIsInstance<Event.Some>()
            .collect { someEvent ->
                doStuff(someEvent)
            }
    }
}.start()

这个想法是在客户端断开连接时始终重新连接（collect 然后返回并开始新的迭代），同时将外部流生命周期与内部（连接）生命周期分开。与潜在的多个订阅者共享流是次要问题。

正如emit 文档所述，它不是线程安全的。那我应该从新的协程中调用它吗？我担心的是，如果外部流没有订阅者，emit 将暂停，无论如何我都需要运行下游管道。

【问题讨论】：

标签： kotlin kotlin-coroutines flow

【解决方案1】：

MutableSharedFlow.emit() 文档说它是is thread-safe。也许你不小心看到了FlowCollector.emit()，它不是线程安全的。 MutableSharedFlow 是 FlowCollector 的子类型，但将 emit() 提升为线程安全的，因为它不像普通的 FlowCollector 那样用作 Flow builder 接收器。没有理由仅仅为了发送到您的共享流而启动协程。

没有理由在使用launch 创建的协程作业上调用start()，因为launch 既创建作业又启动它。

您需要在调用 launch 之前声明 flowToReturn 才能使其在范围内从该外部函数返回。

【讨论】：

是的，我确实看错了emit。 “没有理由仅仅为了发送到您的共享流而启动协程。”暂停发射怎么样？无论流量订阅者的消费能力如何，我都需要收集和doStuff。
好的，我想这是启动协程发射的原因之一。但是随后您将缓冲区大小换成无限数量的协程作业，这些作业更重。替代方法是将 SharedFlow 的 BufferOverflow 策略设置为删除最旧的，或者使用tryEmit()，如果缓冲区已满，这将删除一个项目。
是的，好点，但似乎不可能使用大小不受限制的缓冲区。
@JakubLicznerski 为什么您希望缓冲区实际上不受约束？ MutableSharedFlow<Event>(extraBufferCapacity = Int.MAX_VALUE) 是否足以（并且可能等效于所有意图和目的）以确保即使在共享流上使用慢速收集器也能调用 doStuff？
另外，我猜缓冲区甚至不应该那么大，因为从技术上讲，它应该只用于处理尖峰。如果您的共享流收集器实际上总是比事件的速率慢，那么您最终会耗尽内存（使用launch 时，您会更早耗尽内存，正如@Tenfour04 所指出的那样）。