Kotlin协程：如何合并三个及以上Flow？请求实现zip扩展函数

嘿，好问题！Kotlin Coroutines的标准库里并没有直接提供支持三个及以上Flow的zip函数，不过咱们完全可以基于现有的二元zip来扩展实现，或者写个更通用的版本来满足需求。

先搞定你要的三个Flow的zip扩展 #

直接给你写一个和flow.zip(flow2, flow3) { a, b, c -> }语法完全匹配的扩展函数，用标准库的二元zip链式组合就行：

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.zip

fun <A, B, C, Result> Flow<A>.zip(
    flowB: Flow<B>,
    flowC: Flow<C>,
    transform: suspend (A, B, C) -> Result
): Flow<Result> {
    // 先把前两个Flow zip成Pair，再和第三个Flow zip，最后拆包调用transform
    return this.zip(flowB) { a, b -> Pair(a, b) }
        .zip(flowC) { (a, b), c -> transform(a, b, c) }
}

怎么用？看个例子 #

比如我们有三个不同类型的Flow，这样调用就非常丝滑：

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    val numberFlow = flowOf(1, 2, 3)
    val stringFlow = flowOf("Apple", "Banana", "Cherry")
    val booleanFlow = flowOf(true, false, true)

    val zippedFlow = numberFlow.zip(stringFlow, booleanFlow) { num, fruit, isAvailable ->
        "Item $num: $fruit (Available: $isAvailable)"
    }

    zippedFlow.collect { println(it) }
    // 输出结果：
    // Item 1: Apple (Available: true)
    // Item 2: Banana (Available: false)
    // Item 3: Cherry (Available: true)
}

想要支持更多Flow？ #

如果需要合并四个甚至更多不同类型的Flow，同样可以用链式zip的思路扩展：

fun <A, B, C, D, Result> Flow<A>.zip(
    flowB: Flow<B>,
    flowC: Flow<C>,
    flowD: Flow<D>,
    transform: suspend (A, B, C, D) -> Result
): Flow<Result> {
    return this.zip(flowB, flowC) { a, b, c -> Triple(a, b, c) }
        .zip(flowD) { (a, b, c), d -> transform(a, b, c, d) }
}

要是你要合并的是同类型的Flow，那可以写一个更通用的可变参数版本，支持任意数量：

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow

fun <T, Result> zip(vararg flows: Flow<T>, transform: suspend (List<T>) -> Result): Flow<Result> = flow {
    // 给每个Flow创建迭代器
    val iterators = flows.map { it.iterator() }
    try {
        // 循环收集每个Flow的下一个元素，直到任意一个Flow没有元素为止
        while (true) {
            val elements = iterators.map { it.next() }
            emit(transform(elements))
        }
    } catch (_: NoSuchElementException) {
        // 任意一个Flow结束，整个zip流就结束
    } finally {
        // 确保所有迭代器都关闭
        iterators.forEach { it.close() }
    }
}

注意点 #

别忘了zip的核心特性：它会严格等待所有参与的Flow都发射了对应位置的元素，才会触发一次transform；只要其中任意一个Flow完成了，整个合并后的Flow也会立刻结束。

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