当Kotlin频道变满时执行一段代码

Question

我想写一个简单的批处理器类。它有一个请求队列，等待这个队列变满或者要传递一段时间，然后再与数据库进行对话。

通过通道实现此队列非常方便 - 这样我们的客户端就会在它满了时被挂起。但是我如何才能知道频道是否已满？

当然，我可以创建一个向通道发送内容然后执行一些检查的方法。下一步是将它封装在从Channel派生的类中。还是很脏（目前还不清楚如何处理onSend / onReceive）。有没有更优雅的solutins？也许是开箱即用的东西？

Answer 1

这不是开箱即用的，但是使用actor可以很容易地实现相应的批处理逻辑。你真的不需要一个类（但如果你愿意，你可以将这个代码包装在一个类中）。您可以将以下实现用作模板：

const val MAX_SIZE = 100 // max number of data items in batch
const val MAX_TIME = 1000 // max time (in ms) to wait

val batchActor = actor<Data> {
    val batch = mutableListOf<Data>()
    var deadline = 0L // deadline for sending this batch to DB
    while (true) {
        // when deadline is reached or size is exceeded, then force batch to DB
        val remainingTime = deadline - System.currentTimeMillis()
        if (batch.isNotEmpty() && remainingTime <= 0 || batch.size >= MAX_SIZE) {
            saveToDB(batch)
            batch.clear()
            continue
        }
        // wait until items is received or timeout reached
        select<Unit> {
            // when received -> add to batch
            channel.onReceive {
                batch.add(it)
                // init deadline on first item added to batch
                if (batch.size == 1) deadline = System.currentTimeMillis() + MAX_TIME
            }
            // when timeout is reached just finish select, note: no timeout when batch is empty
            if (batch.isNotEmpty()) onTimeout(remainingTime) {}
        }
    }
}

现在，只要您需要向数据库发送任何内容，您就可以执行batchActor.send(data)，并且actor内部的逻辑负责批处理并将生成的批处理保存到数据库中。

Answer 2

通道接口声明了一个isFull属性，可以查询该属性以确定它是否已达到容量。

我无论如何都看不到有回调函数来在达到容量时自动调用函数但你可以定期检查这个isFull属性以查看它是否处于容量状态。