kotlin - 如何合并 kotlin 协程中的生产者？

Question

使用 Rx 可以合并多个订阅源，如下所示

// psudo data repository

fun getAllData(): Flowable<DataType> {
    return getCachedData().mergeWith(getRemoteData())
}

fun getCachedData(): Flowable<DataType> {
    // local database call
}

fun getRemoteData(): Flowable<DataType> {
    // network call
}

在上面的代码中，getAllData() 将在其中一个合并的 Flowables 返回时立即返回数据，然后在另一个准备好后发送。

问题是，如何使用 Kotlin 协程的 produce 获得相同的结果？

Answer 1

您可以使用 produce 创建一个组合 channel ，您可以在其中启动两个使用两个输入 channel 的协程并将其重新发送到组合 channel 。

这是一个将多个相同类型的接收 channel 合并为一个 channel 的函数。

/**
 * Merges multiple [channels] into one channel.
 * All elements of all channels are sent to the combined channel in the order they arrive on the input channels.
 */
fun <T> CoroutineScope.mergeChannels(vararg channels: ReceiveChannel<T>) : ReceiveChannel<T> {
    return produce {
        channels.forEach {
            launch { it.consumeEach { send(it) }}
        }
    }
}

你可以这样使用它:

fun main() = runBlocking<Unit> {
    val every100Ms = produce {
        repeat(10) {
            send("every 100: $it")
            delay(100)
        }
    }

    val every200Ms = produce {
        repeat(10) {
            send("every 200: $it")
            delay(200)
        }
    }

    val combined = mergeChannels(every100Ms, every200Ms)
    combined.consumeEach { println(it) }
}