运行批量操作作为中间流操作

Question

我有一个 java 未定义长度的流。现在我需要从数据库加载一些元数据并将其分配给流式数据。

我不能：

一次将流中的所有数据加载到我的 RAM，填充元数据，然后启动一个新流，因为这可能会占用大量 RAM。
单独加载每个元素的元数据，因为这会使我的数据库充满太多请求。

因此我想我可以从数据库中加载分区中的元数据。

我需要这样的方法：

<T> Stream<List<T>> partition(Stream<T> stream, int partitionSize)

所以我可以这样使用它

partition(dataSource.stream(), 1000)
    .map(metadataSource::populate)
    .flatMap(List::stream)
    .forEach(this::doSomething);

我已经找到 Guava's Iteralbes#partition 但这会迫使我将流转换为可迭代对象，对其进行分区并再次将其转换为流。流分区是否有内置的东西，或者有一种简单的方法可以自己实现吗？

Answer 1

我还没有找到执行此操作的现有方法，所以我自己实现了一个：

public class Partitioner<E> implements Iterator<List<E>> {

    private final Iterator<E> iterator;
    private final int partitionSize;

    public static <T> Stream<List<T>> partition(final Stream<T> stream, final int partitionSize) {
        return new Partitioner<>(stream, partitionSize).asStream();
    }

    public Partitioner(final Stream<E> stream, final int partitionSize) {
        this(stream.iterator(), partitionSize);
    }

    public Partitioner(final Iterator<E> iterator, final int partitionSize) {
        this.iterator = iterator;
        this.partitionSize = partitionSize;
    }

    @Override
    public boolean hasNext() {
        return this.iterator.hasNext();
    }

    @Override
    public List<E> next() {
        if (!hasNext()) {
            throw new NoSuchElementException("No more elements");
        }
        final ArrayList<E> result = new ArrayList<>(this.partitionSize);
        for (int i = 0; i < this.partitionSize && hasNext(); i++) {
            result.add(this.iterator.next());
        }
        return result;
    }

    public Stream<List<E>> asStream() {
        return StreamSupport.stream(Spliterators.spliteratorUnknownSize(this, Spliterator.NONNULL), false);
    }

}

运行批量操作作为中间流操作

Run bulk operation as intermediate stream operation

java

bulk

java-stream

运行 批量操作作为中间流操作

Run bulk operation as intermediate stream operation

java

bulk

java-stream

运行批量操作作为中间流操作