Kotlin Flow：SharedFlow与StateFlow状态处理的适配难题

配置变更后需保证：

• 数据源响应能被正常消费，不丢失后台请求结果
• 不会重复处理已处理过的内容
• 重复的错误状态能触发UI（比如错误对话框）

当前遇到的矛盾：

• StateFlow：因不发射相同值，重复的错误状态无法触发UI更新
• SharedFlow(replay=1)：配置变更时，旧状态的回放与新请求的结果会被重复收集；移除replay=1则会丢失后台请求结果

方案1：分离持久状态与一次性事件（推荐） #

将需要持久化的UI状态（如资产列表、加载状态）和一次性触发的UI事件（如错误提示、成功Toast）拆分，分别用StateFlow和SharedFlow处理：

1. 定义状态与事件类 #

// 持久化UI状态：配置变更后需保留的状态，用于刷新列表、加载动画等
data class AssetUIState(
    val assetList: List<AssetMinDataDomain>? = null,
    val isLoading: Boolean = false
)

// 一次性UI事件：仅需处理一次的操作，如错误对话框、Toast
sealed class AssetEvent {
    data class ShowError(val msg: String) : AssetEvent()
    object ShowSuccessToast : AssetEvent()
}

2. ViewModel实现 #

class AssetViewModel : ViewModel() {
    // 状态流：保存持久化UI状态，用StateFlow自动回放最新值
    private val _uiState = MutableStateFlow(AssetUIState())
    val uiState: StateFlow<AssetUIState> = _uiState.asStateFlow()

    // 事件流：发送一次性事件，replay=0避免配置变更后回放旧事件
    private val _events = MutableSharedFlow<AssetEvent>(replay = 0, extraBufferCapacity = 1)
    val events: SharedFlow<AssetEvent> = _events.asSharedFlow()

    fun fetchData(param1: Any, param2: Any) {
        viewModelScope.launch {
            // 更新加载状态
            _uiState.update { it.copy(isLoading = true) }
            try {
                val assetList = repository.fetchAssets(param1, param2)
                // 更新成功状态
                _uiState.update {
                    it.copy(
                        assetList = assetList,
                        isLoading = false
                    )
                }
                // 发送成功事件
                _events.emit(AssetEvent.ShowSuccessToast)
            } catch (e: Exception) {
                // 更新失败状态
                _uiState.update { it.copy(isLoading = false) }
                // 发送错误事件（每次错误都是新事件，确保UI触发）
                _events.emit(AssetEvent.ShowError(e.message ?: "未知错误"))
            }
        }
    }
}

3. View层收集逻辑 #

// 收集持久状态：配置变更后自动回放最新状态，UI根据状态刷新列表、加载动画
viewLifecycleOwner.collectState(viewModel.uiState) { state ->
    state.assetList?.let { updateAssetList(it) }
    updateLoadingIndicator(state.isLoading)
}

// 收集一次性事件：仅处理新产生的事件，不会回放旧事件
viewLifecycleOwner.collectEvent(viewModel.events) { event ->
    when (event) {
        is AssetEvent.ShowError -> showErrorDialog(event.msg)
        AssetEvent.ShowSuccessToast -> showSuccessToast()
    }
}

配套扩展函数 #

inline fun <T : Any> LifecycleOwner.collectState(
    stateFlow: StateFlow<T>,
    crossinline action: (T) -> Unit,
    lifecycleState: Lifecycle.State = Lifecycle.State.STARTED
) {
    lifecycleScope.launch {
        repeatOnLifecycle(lifecycleState) {
            stateFlow.collect { action(it) }
        }
    }
}

inline fun <T : Any> LifecycleOwner.collectEvent(
    sharedFlow: SharedFlow<T>,
    crossinline action: (T) -> Unit,
    lifecycleState: Lifecycle.State = Lifecycle.State.STARTED
) {
    lifecycleScope.launch {
        repeatOnLifecycle(lifecycleState) {
            sharedFlow.collect { action(it) }
        }
    }
}

优势 #

• 彻底解决重复处理问题：旧事件不会被回放，仅处理新请求产生的事件
• 重复错误能触发UI：每次错误都是新的AssetEvent实例，确保事件流正常发射
• 配置变更后状态不丢失：StateFlow自动回放最新的持久状态，UI无缝恢复

方案2：给状态添加唯一版本号 #

如果不想拆分状态与事件，可以给AssetState添加版本号，确保即使内容相同，也会被视为新值发射，同时在View层记录已处理的版本号避免重复：

1. 修改状态类 #

sealed class AssetState(val version: Long) {
    data class FetchAssetLoading(
        val assetList: List<AssetMinDataDomain>?,
        override val version: Long
    ) : AssetState(version)
    
    data class FetchAssetSuccess(
        val assetList: List<AssetMinDataDomain>,
        override val version: Long
    ) : AssetState(version)
    
    data class FetchAssetFailed(
        val msg: String,
        val assetList: List<AssetMinDataDomain>?,
        override val version: Long
    ) : AssetState(version)
}

2. ViewModel实现 #

class AssetViewModel : ViewModel() {
    private var stateVersion = 0L
    private val _assetState = MutableStateFlow<AssetState>(
        AssetState.FetchAssetLoading(null, stateVersion++)
    )
    val assetState: StateFlow<AssetState> = _assetState.asStateFlow()

    fun fetchData(param1: Any, param2: Any) {
        viewModelScope.launch {
            // 生成新版本号的加载状态
            val currentList = _assetState.value.currentAssetList()
            _assetState.value = AssetState.FetchAssetLoading(currentList, stateVersion++)
            
            try {
                val assetList = repository.fetchAssets(param1, param2)
                _assetState.value = AssetState.FetchAssetSuccess(assetList, stateVersion++)
            } catch (e: Exception) {
                _assetState.value = AssetState.FetchAssetFailed(
                    e.message ?: "未知错误",
                    currentList,
                    stateVersion++
                )
            }
        }
    }

    // 扩展函数获取当前资产列表
    private fun AssetState.currentAssetList(): List<AssetMinDataDomain>? {
        return when (this) {
            is AssetState.FetchAssetLoading -> assetList
            is AssetState.FetchAssetSuccess -> assetList
            is AssetState.FetchAssetFailed -> assetList
        }
    }
}

3. View层处理逻辑 #

private var lastHandledVersion = -1L

viewLifecycleOwner.collectState(viewModel.assetState) { state ->
    // 仅处理版本号大于已处理的状态
    if (state.version > lastHandledVersion) {
        lastHandledVersion = state.version
        onAssetStateChanged(state)
    }
}

优势 #

• 无需拆分状态与事件，适配原有代码结构
• 重复错误能触发UI：每次状态发射都带新的版本号，StateFlow会正常发射

方案3：取消旧请求避免重复结果 #

在ViewModel中记录当前正在进行的请求，配置变更后重新调用fetchData时，取消旧请求并发起新请求，确保仅处理新请求的结果：

class AssetViewModel : ViewModel() {
    private var currentFetchJob: Job? = null
    private val _assetState = MutableSharedFlow<AssetState>(replay = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
    val assetState: SharedFlow<AssetState> = _assetState.asSharedFlow()

    fun fetchData(param1: Any, param2: Any) {
        // 取消之前的请求，避免旧结果干扰
        currentFetchJob?.cancel()
        currentFetchJob = viewModelScope.launch {
            val currentList = _assetState.replayCache.firstOrNull()?.currentAssetList()
            _assetState.emit(AssetState.FetchAssetLoading(currentList))
            
            try {
                val assetList = repository.fetchAssets(param1, param2)
                _assetState.emit(AssetState.FetchAssetSuccess(assetList))
            } catch (e: Exception) {
                _assetState.emit(AssetState.FetchAssetFailed(e.message ?: "未知错误", currentList))
            }
        }
    }

    private fun AssetState?.currentAssetList(): List<AssetMinDataDomain>? {
        return when (this) {
            is AssetState.FetchAssetLoading -> assetList
            is AssetState.FetchAssetSuccess -> assetList
            is AssetState.FetchAssetFailed -> assetList
            null -> null
        }
    }
}

优势 #

• 确保仅处理新请求的结果，避免旧状态与新结果重复收集
• 适配原有SharedFlow的使用方式

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