kotlin - 如何从 Kotlin 的 Channel 中读取所有可用元素

Question

我想从一个 channel 中读取所有可用的元素，这样如果我的接收器比我的发送器慢，我就可以对它们进行批处理(希望处理批处理的性能更高并允许接收器跟上)。我只想在 channel 为空时暂停，直到我的批次已满或超时才暂停，不像 this question .

标准 kotlin 库中是否内置了任何内容来完成此操作？

Answer 1

我在标准的 kotlin 库中没有找到任何东西，但这是我想出的。这将仅暂停第一个元素，然后是 poll所有剩余的元素。这仅适用于 Buffered Channel以便准备好处理的元素排队并可用于 poll

/**
 * Receive all available elements up to [max]. Suspends for the first element if the channel is empty
 */
internal suspend fun <E> ReceiveChannel<E>.receiveAvailable(max: Int): List<E> {
    if (max <= 0) {
        return emptyList()
    }

    val batch = mutableListOf<E>()
    if (this.isEmpty) {
        // suspend until the next message is ready
        batch.add(receive())
    }

    fun pollUntilMax() = if (batch.size >= max) null else poll()

    // consume all other messages that are ready
    var next = pollUntilMax()
    while (next != null) {
        batch.add(next)
        next = pollUntilMax()
    }

    return batch
}