有没有一种优雅的方式来分块处理流？

Question

我的确切场景是将数据分批插入数据库，所以我想累积 DOM 对象，然后每 1000 个对象刷新一次。

我通过在累加器中放置代码来检测满度然后刷新来实现它，但这似乎是错误的——刷新控制应该来自调用者。

我可以将流转换为 List，然后以迭代方式使用 subList，但这看起来也很笨拙。

是否有一种巧妙的方法来对每 n 个元素采取行动，然后继续处理流，同时只处理一次流？

原文由 Bohemian 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

优雅在旁观者的眼中。如果您不介意在 groupingBy 中使用有状态函数，您可以这样做：

 AtomicInteger counter = new AtomicInteger();

stream.collect(groupingBy(x->counter.getAndIncrement()/chunkSize))
    .values()
    .forEach(database::flushChunk);

这不会比您的原始解决方案赢得任何性能或内存使用点，因为它仍然会在执行任何操作之前具体化整个流。

如果您想避免具体化列表，流 API 将无济于事。您将必须获取流的迭代器或拆分器并执行如下操作：

 Spliterator<Integer> split = stream.spliterator();
int chunkSize = 1000;

while(true) {
    List<Integer> chunk = new ArrayList<>(size);
    for (int i = 0; i < chunkSize && split.tryAdvance(chunk::add); i++){};
    if (chunk.isEmpty()) break;
    database.flushChunk(chunk);
}

原文由 Misha 发布，翻译遵循 CC BY-SA 3.0 许可协议