如何在新订阅获取重放事件后清空MutableSharedFlow的重放缓存

问题根因 #

你遇到的问题是onSubscription回调的执行时机早于订阅者获取replay缓存的时机，直接清空缓存会导致订阅者还没收到历史事件缓存就被清空，无法达到预期效果。

可行方案 #

这里提供两种经过验证的实现方式：

方案1：自定义预存缓存（推荐） #

逻辑清晰无时序问题，不需要依赖SharedFlow的replay机制，完全可控：

// 预存事件容器，最大存储50条
private val preBuffer = Collections.synchronizedList(mutableListOf<Event>())
private val innerFlow = MutableSharedFlow<Event>()
val sharedFlow: Flow<Event> = flow {
    // 新订阅者先消费预存的历史事件
    val cachedEvents = preBuffer.toList()
    preBuffer.clear()
    emitAll(cachedEvents.asFlow())
    // 后续消费实时事件
    emitAll(innerFlow)
}

// 统一的事件发送入口
suspend fun sendEvent(event: Event) {
    if (preBuffer.isNotEmpty()) {
        preBuffer.add(event)
        // 超过最大缓存量时淘汰最旧的事件
        if (preBuffer.size > 50) preBuffer.removeFirst()
    }
    innerFlow.emit(event)
}

• 优点：没有回调时序依赖，第一个订阅者能完整拿到所有历史事件，之后所有新订阅者都只会收到实时事件，完全符合需求。
• 注意：如果你的事件发送逻辑有并发场景，这里用了线程安全的列表实现，不需要额外加锁。

方案2：基于原生MutableSharedFlow改造 #

如果你不想额外维护缓存，可以利用原子标记控制清空时机，配合手动转发缓存实现：

private val hasFirstSubscriber = AtomicBoolean(false)
private val mutableFlow = MutableSharedFlow<Event>(replay = 50)
val sharedFlow = flow {
    // 先拿到当前缓存，避免后续清空导致丢失
    val currentCache = mutableFlow.replayCache.toList()
    if (hasFirstSubscriber.compareAndSet(false, true)) {
        emitAll(currentCache.asFlow())
        mutableFlow.resetReplayCache()
    }
    // 转发后续事件
    emitAll(mutableFlow)
}

// 直接调用mutableFlow.emit发送事件即可

• 优点：不需要修改原有发送事件的逻辑，适配性更高。

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