将多个流合并并转换为单个流？答案

【问题标题】：Combine and transform multiple flows into a single flow?将多个流合并并转换为单个流？
【发布时间】：2022-01-11 00:52:26
【问题描述】：

我正在尝试开发一些代码，允许 N 个生产者产生值并让单个消费者消费所有流并在其中一个/多个流发生变化时运行特定方法。例如，我的消费者处理由许多不同流发出的字符串值的排序/排序。

制作人A：

private val _someFlow = MutableStateFlow("")
val someFlow = _someFlow 

fun someMethod(stringToEmit: String) {
    _someFlow.emit(stringToEmit) //wrapped in a coroutine
}

生产者B：

private val _someFlow = MutableStateFlow("")
val someFlow = _someFlow

fun someMethod(stringToEmit: String) {
    _someFlow.emit(stringToEmit) //wrapped in a coroutine
}

对任意 N 个流重复上述模式。

我如何将来自所有生产者的 N 个流合并到一个消费者中，在那里我可以做一些事情，比如对所有这些流进行排序并输出已排序流的列表？

看起来 Flow 具有 combine 函数，看起来与我想要的相似，但是当其中一个或多个流发生变化时，我想将一些函数应用于所有流。如果流没有改变，那么就使用它的旧值。

似乎这样的事情可能会起作用：

val sorted = producerA.someFlow.combine(producerB.someFlow) { producerA, producerB -> 
    return sort(producerA, producerB)
}

但是如果我有 N 个流，上面的 combine 将如何扩展以支持它？这是 Flow 以外的其他东西的用例吗？我想灵活地允许消费者在他们想要的任何协程范围内向他们的流发出。

【问题讨论】：

combine 有一个可变参数版本，但如果您的流属于不同类型，则必须在 lambda 中转换参数。
@Tenfour04 它们将具有相同的类型。我想我也很好奇combine 的可变参数版本中是否需要明确的 lambda 参数？ IE。如果我有 8 个要组合的流，我需要在 lambda 中定义 8 个参数吗？
这里是它的所有变体。对于可变参数版本，lambda 参数是一个数组。对于五个或更少的流，存在特定的重载，您可以在其中获取具有特定类型的所有参数。 kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/…

标签： kotlin kotlin-coroutines flow kotlin-flow

【解决方案1】：

如果您的生产者流具有相同的类型，则有一个方便的combine 函数，它接受未定义数量的流：

inline fun <T, R> combine(vararg flows: Flow<T>, crossinline transform: suspend (Array<T>) -> R): Flow<R>

它将每个流最近发出的值组合到Array。示例：

combine(someFlow1, someFlow2, someFlow3, someFlow4) { result: Array<String> ->
    //... use result
}

【讨论】：