kotlin - 如何使用 Kotlin channel 实现自然(又名智能)批处理？

Question

自然又名。智能批处理是流处理中的一种技术，可在不影响延迟的情况下优化吞吐量。在并发队列的例子中，消费者有能力自动地排出在某个时刻观察到的所有项目，然后将它们作为批处理进行处理。理想情况下，队列应该是有界的，给批处理大小一个上限，同时为发送方提供背压。

之所以称为“自然”批处理，是因为没有强加的批处理大小:当流量较低时，它会在每个项目到达时立即对其进行处理。在这种情况下，您不需要通过将项目一起批处理来进行任何吞吐量优化。当流量变高时，消费者将自动开始处理更大的批次，分摊单个操作(如数据库)的固定延迟 INSERT .

我写了一些实现基本目标的代码:

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

const val batchLimit = 20

@ObsoleteCoroutinesApi
suspend inline fun <T: Any> ReceiveChannel<T>.consumeBatched(
        handleItems: (List<T>) -> Unit
) {
    val buf = mutableListOf<T>()
    while (true) {
        receiveOrNull()?.also { buf.add(it) } ?: break
        for (x in 2..batchLimit) {
            poll()?.also { buf.add(it) } ?: break
        }
        handleItems(buf)
        buf.clear()
    }
}

我们可以用这个来测试它:

@ObsoleteCoroutinesApi
@ExperimentalCoroutinesApi
fun main() {
    val chan = generateMockTraffic()
    runBlocking {
        chan.consumeBatched { println("Received items: $it") }
    }
}

@ExperimentalCoroutinesApi
private fun generateMockTraffic(): ReceiveChannel<Int> {
    return GlobalScope.produce(capacity = batchLimit) {
        (1..100).forEach {
            send(it)
            if (it % 10 == 0) {
                delay(1)
            }
        }
    }
}

consumeBatched()一次轮询队列一项，因此必须另外施加批处理限制。如果针对 Agrona 项目的 OneToOneConcurrentArrayQueue 之类的并发队列进行写入会更好。 , 支持 drain手术。

Kotlin channel 是否有更好的方法，并获得库的更多支持？

如果没有，这是否会被视为要添加的功能？

Answer 1

Is there a better approach with Kotlin channels, with more support from the library?

该库不支持此功能。

If not, would this be considered as a feature to add?

这取决于所需的 API 表面。 drain成员不太可能适合 channel 语义:它限制了实现，它应该以某种方式公开消耗限制，并为 channel 提供更多“类似集合”的 API。例如。应该怎么做 drain使用无限 channel ？是否可以实现 drain以有效的方式(使用预先调整大小的缓冲区，但避免 OOM 和无限集合) 一次 并将其与任何 channel 实现一起使用？

可以改进的是来自 channel 的额外提示，例如预期容量和排队元素的数量。它们可以具有默认实现的宽松语义，并像对 drain 的提示一样起作用。扩展具有一些合理的可配置上限。以后可以添加此类API，请随时联系 create an issue