如何合并多个Flow，避免单个Flow的debounce操作影响其他Flow？

需求 #

我有两个数据流：

• intFlow：更新频率低，要求实时响应，不受延迟操作影响
• stringFlow：数据波动大，需要用debounce(2000)降速，仅在2秒无更新时触发合并

需要将两者合并后传递给UI，要求intFlow更新时立即触发合并，stringFlow稳定后再触发合并。

尝试过的方案及问题 #

1. combine()：intFlow会等待stringFlow的数据，无法实现intFlow实时响应的需求
2. flatMapMerge{}：intFlow的数据接收存在约2秒延迟，输出结果不符合预期
3. merge()取巧写法：通过类型判断直接访问StateFlow.value，存在并发风险（可能拿到过期数据），代码如下：

.map {
    if (it is Int) Pair(stringFlow.value, it) else Pair(it as String, intFlow.value)
}

正确实现方案 #

核心思路：每次触发合并事件（intFlow更新或stringFlow稳定后），通过combine实时获取两个流的最新值，避免直接访问value带来的并发问题。

修正后的transform函数 #

fun transform(result: suspend (s: String, i: Int) -> Flow<String>): Flow<String> {
    val debouncedStringFlow = stringFlow.debounce(2000)
    
    // 合并触发源：intFlow实时触发，debounce后的stringFlow延迟触发
    return merge(intFlow, debouncedStringFlow)
        // 每次触发时，实时获取两个流的最新组合值
        .flatMapLatest {
            combine(stringFlow, intFlow) { s, i -> Pair(s, i) }
                .take(1) // 仅取当前最新的一次组合
        }
        .filter { (s, i) -> s.isNotBlank() && i > 0 }
        .distinctUntilChanged()
        .flatMapLatest { (s, i) -> result(s, i) }
}

完整测试代码 #

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import java.time.LocalDateTime
import java.time.ZoneId
import java.time.format.DateTimeFormatter

val intFlow = MutableStateFlow(0)
val stringFlow = MutableStateFlow("")

fun transform(result: suspend (s: String, i: Int) -> Flow<String>): Flow<String> {
    val debouncedStringFlow = stringFlow.debounce(2000)
    
    return merge(intFlow, debouncedStringFlow)
        .flatMapLatest {
            combine(stringFlow, intFlow) { s, i -> Pair(s, i) }
                .take(1)
        }
        .filter { (s, i) -> s.isNotBlank() && i > 0 }
        .distinctUntilChanged()
        .flatMapLatest { (s, i) -> result(s, i) }
}

val combinedFlow = transform { s, i ->
    flow {
        emit("$s::$i")
    }
}

fun getCurrentTime(): String {
    val currentDateTime = LocalDateTime.now()
    val formatter = DateTimeFormatter.ofPattern("HH:mm:ss:SSS")
    return currentDateTime.format(formatter) + " | ${
        currentDateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli()
    }"
}

suspend fun main() {
    val mainJob = CoroutineScope(Dispatchers.Default).launch {
        println("${getCurrentTime()} - Updating String to 'a'")
        stringFlow.update { "a" }

        println("${getCurrentTime()} - Updating Int to 1")
        intFlow.update { 1 }

        println("${getCurrentTime()} - Updating String to 'a'")
        stringFlow.update { "a" }

        println("${getCurrentTime()} - Updating Int to 2")
        intFlow.update { 2 }

        println("${getCurrentTime()} - Delaying for 5 seconds")
        delay(5000)

        println("${getCurrentTime()} - Updating String with 'b'")
        stringFlow.update { "b" }

        println("${getCurrentTime()} - Updating Int to 3")
        intFlow.update { 3 }
    }

    val collectJob = CoroutineScope(Dispatchers.Default).launch {
        combinedFlow.collect {
            println("${getCurrentTime()} - Combined: $it-------- ")
        }
    }

    mainJob.join()
    collectJob.join()
}

预期输出 #

21:XX:XX:XXX - Updating String to 'a'
21:XX:XX:XXX - Updating Int to 1
21:XX:XX:XXX - Combined: a::1-------- 
21:XX:XX:XXX - Updating String to 'a'
21:XX:XX:XXX - Updating Int to 2
21:XX:XX:XXX - Combined: a::2-------- 
21:XX:XX:XXX - Delaying for 5 seconds
21:XX:XX:XXX - Updating String with 'b'
21:XX:XX:XXX - Updating Int to 3
21:XX:XX:XXX - Combined: b::3-------- 
21:XX:XX:XXX - Combined: b::3-------- 

方案优势 #

• 完全避免直接访问StateFlow.value的并发风险，所有值获取都在Flow的上下文内完成
• intFlow更新时立即触发合并，实时拿到最新的string和int值
• stringFlow仅在稳定2秒后触发合并，同时获取最新的int值，不会出现延迟

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