Kotlin如何将Sequence<T>按N个元素分割为Sequence<Sequence<T>>？

数据老陈

2026-8-26

问题

我正在开发一个高负载Kotlin应用，需要将数万条数据批量插入数据库（例如每批1000条）。我编写了如下循环代码：

val itemsSeq = itemsList.iterator().asSequence()
while (true) {
    log.debug("Taking $BATCH_SIZE from $itemsSeq")
    val batchSeq = itemsSeq.take(BATCH_SIZE)
    val squareBatch = applySomething(batchSeq, something)
        ?: break
}

fun applySomething(batch: Sequence<Item>, something: Something) {
    /* Fully consumes batch. Bulk-loads from DB by IDs, applies, bulk-saves. */ 
}

我原本认为take()会推进itemsSeq，下一次调用take()会获取从第N+1个元素开始的Sequence视图，但运行时出现如下异常：

DEBUG Taking 10 from kotlin.sequences.ConstrainedOnceSequence@53fe15ff
Exception in thread "main" java.lang.IllegalStateException: This sequence can be consumed only once.
    at kotlin.sequences.ConstrainedOnceSequence.iterator(SequencesJVM.kt:23)
    at kotlin.sequences.TakeSequence$iterator$1.<init>(Sequences.kt:411)
    at kotlin.sequences.TakeSequence.iterator(Sequences.kt:409)

可见take()会再次"打开"仅能单次消费的itemsSeq，导致异常。目前我可以使用chunked()方法作为替代：

public fun <T> Sequence<T>.chunked(size: Int): Sequence<List<T>> {

但我希望避免生成List，而是直接得到Sequence<Sequence<T>>，即介于take()和chunked()之间的功能。请问Kotlin SDK中是否有此类内置方法？我可以自行实现sequence{...}但更倾向于使用可读性更高的内置方案。

解决方案

Kotlin标准库目前没有直接提供返回Sequence<Sequence<T>>的分批次内置方法，不过可以通过两种方式满足需求：

自定义序列分批次扩展方法
利用sequence构建器封装迭代器的分批逻辑，既避免生成中间List，又保证单次消费原序列：

fun <T> Sequence<T>.batchSequence(size: Int): Sequence<Sequence<T>> = sequence {
    val iterator = this@batchSequence.iterator()
    while (iterator.hasNext()) {
        yield(sequence {
            repeat(size) {
                if (iterator.hasNext()) {
                    yield(iterator.next())
                } else {
                    return@sequence
                }
            }
        })
    }
}

使用时直接调用该扩展方法遍历批次序列即可：

itemsSeq.batchSequence(BATCH_SIZE).forEach { batch ->
    applySomething(batch, something)
}

适配现有chunked方法
如果不想自定义扩展，也可以调整applySomething的输入参数，或者将chunked生成的List转为Sequence使用——虽然会生成中间List，但Kotlin的chunked实现已做优化，多数高负载场景下性能开销可接受：

itemsSeq.chunked(BATCH_SIZE).forEach { batchList ->
    applySomething(batchList.asSequence(), something)
    // 或者直接修改applySomething参数为List<Item>
}

需要注意：原代码中itemsList.iterator().asSequence()生成的是ConstrainedOnceSequence，这类序列只能被迭代一次，多次调用take会尝试重新获取迭代器，从而抛出异常。而自定义的batchSequence会持有唯一的迭代器，保证每次批次消费都推进原序列的位置。

内容的提问来源于stack exchange，提问作者Ondra Žižka