我想从一个 channel 中读取所有可用的元素,这样如果我的接收器比我的发送器慢,我就可以对它们进行批处理(希望处理批处理的性能更高并允许接收器跟上)。我只想在 channel 为空时暂停,直到我的批次已满或超时才暂停,不像 this question .
标准 kotlin 库中是否内置了任何内容来完成此操作?
最佳答案
我在标准的 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
}
关于kotlin - 如何从 Kotlin 的 Channel 中读取所有可用元素,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/49618652/