Android多源更新Token时，Kotlin Flow方案选型与StateFlow疑问

场景描述 #

在Android项目中，token属性会被两个远程数据源更新，预期只处理最新的更新或与前一次值不同的更新。目前正在学习Kotlin Flow，尝试了两种实现方式，想请教哪种更适配场景，同时疑惑StateFlow如何判断哪个emit是最新的。

方式一：结合distinctUntilChanged()与collectLatest() #

object TokenManager {
    private val tokenFlow = MutableStateFlow<String?>(null)

    // 暴露合并双数据源且去重的Flow
    val combinedTokenFlow = tokenFlow
            .distinctUntilChanged()
            .catch { Log.e("TokenManager", "获取Token失败", it) }

    fun updateToken(token: String?) {
        tokenFlow.tryEmit(token)
    }
}

// 数据源1：监听Token更新的服务
class TokenListenerService : TokenService() {
    override fun onTokenUpdate(s: String) {
        TokenManager.updateToken(s)
    }
}

// 数据源2：另一个触发Token变更的逻辑
fun init() {
    // 设置Token监听器
    val tokenListener = OnCompleteListener<String?> { token ->
        TokenManager.updateToken(token)
    }

    // 注册监听器
    Messaging.getInstance().token.addOnCompleteListener(tokenListener)

    // 开始监听Flow中的Token更新
    startObservingToken()
}

fun startObservingToken() {
    launch {
        TokenManager.combinedTokenFlow // 两个数据源的更新都会流入这里
            .collectLatest { token -> // 使用collectLatest
                // 处理Token更新逻辑
            }
    }
}

方式二：结合distinctUntilChanged()、conflate()与collect() #

object TokenManager {
    private val tokenFlow = MutableStateFlow<String?>(null)

    // 暴露合并双数据源且去重的Flow
    val combinedTokenFlow = tokenFlow
            .distinctUntilChanged()
            .conflate() // 只保留最新值
            .catch { Log.e("TokenManager", "获取Token失败", it) }

    fun updateToken(token: String?) {
        tokenFlow.tryEmit(token)
    }
}

// 监听逻辑部分
fun startObservingToken() {
    launch {
        TokenManager.combinedTokenFlow // 两个数据源的更新都会流入这里
            .collect { token -> // 使用collect
                // 处理Token更新逻辑
            }
    }
}

更新后的实现（基于评论优化） #

interface TokenProvider {
    val combinedTokenFlow: Flow<String?>
    fun updateToken(token: String?)
}

class TokenManager : TokenProvider {
    private val _tokenFlow = MutableStateFlow<String?>(null)

    // 暴露合并双数据源且去重的Flow
    override val combinedTokenFlow: Flow<String?>
        get() = _tokenFlow

    // 更新Token的方法
    override fun updateToken(token: String?) {
        _tokenFlow.tryEmit(token)
    }
}

val tokenProvider: TokenProvider = TokenManager()

// 数据源1：监听Token更新的服务
class TokenListenerService : TokenService() {
    override fun onTokenUpdate(s: String) {
        TokenManager.updateToken(s)
    }
}

class NonUiModule : CoroutineScope {
    private val job = SupervisorJob()
    override val coroutineContext = Dispatchers.Default + job

    fun init() {
        // 设置Token监听器
        val tokenListener = OnCompleteListener<String?> { token ->
            TokenManager.updateToken(token)
        }

        // 注册监听器
        Messaging.getInstance().token.addOnCompleteListener(tokenListener)

        // 开始监听Flow中的Token更新
        startObservingToken()
    }

    fun startObservingToken() {
        launch {
            TokenManager.combinedTokenFlow // 两个数据源的更新都会流入这里
                .catch { Log.e("NonUiModule", "获取Token失败", it) }
                .collect { token -> // 使用collect
                    doProcessUpdate(token)
                }
        }
    }

    // ... 其他方法 ...
}

疑问 #

不确定StateFlow如何判定哪个emit是最新的（因为更新可能随时来自不同的数据源），想请教哪种方案更适配该场景？

回答 #

StateFlow的"最新值"判定逻辑

StateFlow是有状态的热流，它会始终保留当前最新的状态值。不管更新来自哪个数据源，tryEmit的调用顺序就是判定"最新"的依据——后调用tryEmit的值会覆盖之前的，成为当前最新状态。如果两个数据源几乎同时调用tryEmit，协程调度器的执行顺序决定最终的最新值。

两种方案的适配性分析

1. 方式一（distinctUntilChanged + collectLatest）

   • 核心逻辑：distinctUntilChanged确保只发送与前一次不同的Token；collectLatest会取消当前正在执行的处理逻辑，立即处理最新的Token。
   • 适配场景：如果Token处理逻辑耗时较长（比如发起网络请求同步Token），且只关心最终的最新Token结果，不需要处理中间值，这种方案更合适。比如短时间内Token连续更新3次，collectLatest会跳过前两次的处理，直接处理第三次的最新值。

2. 方式二（distinctUntilChanged + conflate + collect）

   • 核心逻辑：conflate会让Flow只保留最新的未处理值，当前处理逻辑完成后直接处理最新值；collect会按顺序处理每个值，但中间被覆盖的旧值会被跳过。
   • 适配场景：如果Token处理逻辑耗时较短，或者希望确保每一个"有效更新"（与前值不同的更新）最终都被处理，但可以跳过中间被覆盖的重复更新，这种方案更稳妥。比如短时间内Token连续更新3次，conflate会保留第三次的值，等第一次处理完成后直接处理第三次，跳过第二次。

针对场景的最优建议

结合需求（处理最新更新/与前值不同的更新），两种方案都能满足基础需求，可根据处理逻辑的特性选择：

• 若不需要处理中间更新过程，只想要最终最新结果，方式一更高效，避免无用的中间处理。
• 若处理逻辑轻量，且希望尽可能处理每一次有效更新，方式二更稳妥。

另外，更新后的实现用TokenProvider接口做抽象是很好的实践，建议后续统一通过tokenProvider实例调用，避免直接依赖具体实现，提升可测试性。

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