Kotlin SharedFlow 合并操作在特定情况下具有 zip 行为

问题描述

我正在组合两个 SharedFlows，然后执行长时间的工作操作。

一开始，我知道状态，所以我为两个流发出一个“起始值”。之后，用户可以向任一流发送。

两个流大多是独立的，但在特定情况下，用户可以同时向两个流发送。这样做的作用是组合被触发两次并且长时间的工作被执行两次，实际上，在这种情况下，我只对接收两个值感兴趣但只执行一次工作。

这是我所拥有的：

val _numbers = MutableSharedFlow<Int>(replay = 0,extraBufferCapacity = 1,onBufferOverflow = BufferOverflow.DROP_OLDEST)
val numbers: SharedFlow<Int> = _numbers
val _strings = MutableSharedFlow<String>(replay = 0,onBufferOverflow = BufferOverflow.DROP_OLDEST)
val strings: SharedFlow<String> = _strings

combine(numbers,strings) { (number,strings) ->
    println("values $number - $strings. Starting to perform a long working job")
}
    .launchIn(Coroutinescope(dispatchers.IO))

runBlocking {

    delay(500)
    // This is the initial values. I always kNow this at start.
    _numbers.emit(0)
    _strings.emit("a")

    // Depending of user action,number or string is emitted.

    delay(100)
    _numbers.emit(1)
    delay(100)
    _numbers.emit(2)
    delay(100)
    _numbers.emit(3)
    delay(100)
    _numbers.emit(4)
    delay(100)
    _strings.emit("b")
    delay(100)
    _strings.emit("c")
    delay(100)
    _strings.emit("d")
    delay(100)
    _strings.emit("e")
    delay(100)

    // In a specific situation both values need to change but I only want to trigger the long working job once
    _numbers.emit(10)
    _strings.emit("Z")
}

这可以产生这个：

values 0 - a. Starting to perform a long working job
values 1 - a. Starting to perform a long working job
values 2 - a. Starting to perform a long working job
values 3 - a. Starting to perform a long working job
values 4 - a. Starting to perform a long working job
values 4 - b. Starting to perform a long working job
values 4 - c. Starting to perform a long working job
values 4 - d. Starting to perform a long working job
values 4 - e. Starting to perform a long working job
values 10 - e. Starting to perform a long working job
values 10 - Z. Starting to perform a long working job

或者这个：

values 0 - a. Starting to perform a long working job
values 1 - a. Starting to perform a long working job
values 2 - a. Starting to perform a long working job
values 3 - a. Starting to perform a long working job
values 4 - a. Starting to perform a long working job
values 4 - b. Starting to perform a long working job
values 4 - c. Starting to perform a long working job
values 4 - d. Starting to perform a long working job
values 4 - e. Starting to perform a long working job
values 10 - Z. Starting to perform a long working job

由于缓冲区溢出，有时我可以实现我想要的（最新的），但在其他情况下，我对 values 10 - e. Starting to perform a long working job 不感兴趣。

有什么办法可以强制执行，当发送给两个时，只开始一次长时间的工作？

https://pl.kotl.in/JA1Wdhra9

解决方法

如果你想保持 2 个流，单事件和双事件之间的区别必须是基于时间的。您将无法区分字符串然后数字的快速更新和“双重更新”。

如果基于时间对您来说没问题，那么在长时间处理之前使用 debounce 应该是可行的方法：

combine(numbers,strings) { (number,string) -> number to string }
    .debounce(50)
    .onEach { (number,string) ->
        println("values $number - $string. Starting to perform a long working job")
    }
    .launchIn(CoroutineScope(Dispatchers.IO))

此处，combine 仅从 2 个流构建对，但仍获取所有事件，然后 debounce 忽略快速连续事件并仅发送快速系列中的最新事件。这也带来了轻微的延迟，但这完全取决于您想要实现的目标。

如果基于时间的区分不适合您，您需要一种让生产者以不同于 2 个单事件的方式发送双事件的方法。为此，您可以使用单个事件流，例如，您可以像这样定义事件：

sealed class Event {
    data class SingleNumberUpdate(val value: Int): Event()
    data class SingleStringUpdate(val value: String): Event()
    data class DoubleUpdate(val num: Int,val str: String): Event()
}

但是你必须自己编写“组合”逻辑（保持最新数字和字符串的状态）：

flow {
    var num = 0
    var str = "a"
    emit(num to str)
    events.collect { e ->
        when (e) {
            is Event.SingleNumberUpdate -> {
                num = e.value
            }
            is Event.SingleStringUpdate -> {
                str = e.value
            }
            is Event.DoubleUpdate -> {
                num = e.num
                str = e.str
            }
        }
        emit(num to str)
    }
}
.onEach { (number,strings) ->
    println("values $number - $strings. Starting to perform a long working job")
}
.launchIn(CoroutineScope(Dispatchers.IO))

kotlin kotlin kotlin kotlin-sharedflow