Kotlin中合并多个同类型Channel的优雅实现方案咨询

你可以利用Kotlin协程的select表达式或多协程转发的方式来实现需求，这两种方案都能避免原实现中的忙等空转问题，让协程在无数据时挂起，高效利用资源。

方案一：多协程转发（简洁易读） #

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun <T> join(channel1: ReceiveChannel<T>, channel2: ReceiveChannel<T>): ReceiveChannel<T> = produce {
    val channels = listOf(channel1, channel2)
    var activeCount = channels.size

    // 为每个输入channel启动单独协程转发数据
    for (channel in channels) {
        launch {
            try {
                // 遍历channel所有数据并转发
                for (item in channel) {
                    send(item)
                }
            } finally {
                // 单个channel关闭后，活跃数减1；全部关闭时关闭输出channel
                if (--activeCount == 0) {
                    close()
                }
            }
        }
    }
}

方案二：select表达式精准控制 #

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun <T> join(channel1: ReceiveChannel<T>, channel2: ReceiveChannel<T>): ReceiveChannel<T> = produce {
    var isChannel1Active = true
    var isChannel2Active = true

    while (isChannel1Active || isChannel2Active) {
        select {
            // 监听channel1的数据接收或关闭事件
            if (isChannel1Active) {
                channel1.onReceive { item ->
                    send(item)
                }
                channel1.onReceiveCatching { result ->
                    isChannel1Active = false
                    // 处理channel关闭前的最后一条数据
                    result.getOrNull()?.let { send(it) }
                }
            }
            // 监听channel2的数据接收或关闭事件
            if (isChannel2Active) {
                channel2.onReceive { item ->
                    send(item)
                }
                channel2.onReceiveCatching { result ->
                    isChannel2Active = false
                    result.getOrNull()?.let { send(it) }
                }
            }
        }
    }
}

原方案的问题分析 #

你之前用tryReceive循环的实现属于忙等操作：协程会持续占用CPU循环检查channel状态，不会挂起等待数据，在无数据时完全是资源浪费。而上面的两种方案都会在无数据时挂起协程，直到有数据可读或channel关闭，性能和资源利用率更优。

使用示例 #

fun main() = runBlocking {
    val channel1 = produce {
        repeat(3) { send("Channel1 数据: $it") }
        close()
    }
    val channel2 = produce {
        repeat(3) { send("Channel2 数据: $it") }
        close()
    }

    val joinedChannel = join(channel1, channel2)
    for (item in joinedChannel) {
        println(item)
    }
    println("所有数据接收完成")
}

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