有没有一种优雅的方法可以分块处理流？

java java-8 java-stream chunking

2022-08-31 14:02:03

我的确切方案是将数据批量插入数据库，因此我想累积DOM对象，然后每1000个刷新它们。

我通过将代码放入累加器中以检测完整性然后刷新来实现它，但这似乎是错误的 - 刷新控件应来自调用方。

我可以将流转换为列表，然后以迭代方式使用subList，但这似乎也很笨拙。

有没有一种简洁的方式来采取行动，然后继续处理流，同时只处理一次流？

答案 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);
}

答案 2

上面的大多数答案都没有使用流的好处，比如节省你的内存。您可以尝试使用迭代器来解决问题

Stream<List<T>> chunk(Stream<T> stream, int size) {
  Iterator<T> iterator = stream.iterator();
  Iterator<List<T>> listIterator = new Iterator<>() {

    public boolean hasNext() {
      return iterator.hasNext();
    }

    public List<T> next() {
      List<T> result = new ArrayList<>(size);
      for (int i = 0; i < size && iterator.hasNext(); i++) {
        result.add(iterator.next());
      }
      return result;
    }
  };
  return StreamSupport.stream(((Iterable<List<T>>) () -> listIterator).spliterator(), false);
}