Kotlin SharedFlow 组合操作。在特定情况下有 zip 行为

Question

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

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

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

这是我的:

val _numbers = MutableSharedFlow<Int>(replay = 0, extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
val numbers: SharedFlow<Int> = _numbers
val _strings = MutableSharedFlow<String>(replay = 0, extraBufferCapacity = 1, 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

由于缓冲区溢出，有时我可以实现我想要的(最新的)但在其他情况下，我有 值 10 - e。开始执行我不感兴趣的长时间工作。

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

https://pl.kotl.in/JA1Wdhra9

Answer 1

如果要保留 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))