如何让Kotlin Flow的onEach仅执行一次并被下游两个收集器接收

核心原因 #

你定义的Repository.sharedFlow是一条冷流处理链——每次有新订阅者（比如EventCollectorA和EventCollectorB）订阅时，都会从头执行一遍filter、onEach等变换逻辑。两个订阅者就会触发两次onEach的执行，这就是问题的根源。

解决方案 #

给Repository的流加上shareIn操作符，把这条冷流链转换成共享热流。这样上游的处理逻辑（包括onEach）只会执行一次，处理后的事件会分发给所有订阅者。

修改后的Repository代码：

class Repository constructor(
    eventHandler: EventHandler,
    private val globalScope: CoroutineScope // 传入一个全局稳定的CoroutineScope
) {
    val sharedFlow = eventHandler.sharedFlow
            .filter { it == "Click Event" }
            .onEach {/* 这里的逻辑现在只会执行一次 */}
            .onStart { emit("Begin") }
            .shareIn(
                scope = globalScope,
                started = SharingStarted.WhileSubscribed(5000), // 无订阅时延迟5秒关闭，节省资源
                replay = 1 // 新订阅者会收到最新的1条历史事件，可按需调整
            )
}

关键细节说明 #

1. CoroutineScope的选择：必须用一个生命周期长于所有订阅者的Scope，比如全局单例Scope，不能用ViewModel的viewModelScope（ViewModel销毁后流会停止）。
2. SharingStarted策略：
   • WhileSubscribed(5000)：推荐使用，当有订阅者时保持活跃，无订阅后延迟5秒关闭，平衡资源占用和响应速度。
   • Eagerly：立即启动并一直保持活跃，适合需要持续监听的场景。
3. replay参数：控制新订阅者订阅时能拿到的历史事件数量，比如设为0就不会重播历史事件。

附：用Hilt提供全局CoroutineScope #

如果用Hilt管理依赖，可以添加一个全局Scope的提供类：

@Module
@InstallIn(SingletonComponent::class)
object CoroutineScopeModule {
    @Provides
    @Singleton
    fun provideGlobalCoroutineScope(): CoroutineScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
}

然后修改Repository的构造函数，让Hilt自动注入这个Scope：

class Repository @Inject constructor(
    eventHandler: EventHandler,
    private val globalScope: CoroutineScope
) {
    // ... 流定义不变
}

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