Stream.generate(s).limit(n) 是否保证对生成器函数 s 进行 n 次调用，或者是否有更好的替代方案？

Question

我有一个数据源，我知道它有

n

元素，我可以通过重复调用对象上的方法来访问它。 为了举例，我们称其为

myReader.read()

。 我想创建一个包含这些

n

元素的数据流。 还可以说，我不想调用

read()

方法的次数超过我想要返回的数据量，因为如果在该方法结束后调用该方法，它将抛出异常（例如

NoSuchElementException

）数据已达到。

我知道我可以使用

IntStream.range

方法创建此流，并使用

read

方法映射每个元素。 然而，这感觉有点奇怪，因为我完全忽略了流中的 int 值（我实际上只是用它来生成一个恰好包含

n

元素的流）。

Stream<String> myStream =
        IntStream.range(0, n).mapToObj(i -> myReader.read());

我考虑过的一种方法是使用

Stream.generate(supplier)

，然后使用

Stream.limit(maxSize)

。 根据我对

limit

函数的理解，这感觉应该可行。

Stream<String> myStream = Stream.generate(myReader::read).limit(n)

但是，我在 API 文档中没有看到任何迹象表明

Stream.limit()

方法将保证

maxSize

元素由调用它的流准确生成。 只要最终结果只是第一个

n

调用，并且只要它满足 API 契约，那么允许流实现调用生成器函数超过

n

次就不是不可行的。是一个短路中间操作。

Stream.limit JavaDocs

返回由该流的元素组成的流，长度被截断为不超过 maxSize。 这是一个短路状态中间操作。

流操作和管道文档

如果在提供无限输入时，中间操作可能会产生有限流，则中间操作是短路的。 [...] 管道中的短路操作是无限流处理在有限时间内正常终止的必要条件，但不是充分条件。

仅依赖

Stream.generate(generator).limit(n)

对底层生成器进行

n

调用是否安全？ 如果是这样，是否有一些我缺少的事实证明文件？

为了避免 XY 问题：通过精确执行操作 n 次来创建流的惯用方法是什么？

Answer 1

Stream.generate

创建一个

无序流。这意味着后续的 limit 操作不需要使用前

n

元素，因为当没有顺序时没有“第一个”，但可以选择任意 n 元素。实现可能会利用此权限，例如以获得更高的并行处理性能。 以下代码

IntSummaryStatistics s =
    Stream.generate(new AtomicInteger()::incrementAndGet)
        .parallel()
        .limit(100_000)
        .collect(Collectors.summarizingInt(Integer::intValue));

System.out.println(s);

打印类似的东西

IntSummaryStatistics{count=100000, sum=5000070273, min=1, average=50000,702730, max=100207}

在我的机器上，而最大数量可能会有所不同。它表明 Stream 已根据需要准确选择了 

100000

个元素，但没有选择 1 到 100000 之间的元素。由于生成器生成严格升序的数字，因此很明显，它已被调用超过 100000 次才能获得高于该数字的数字.

另一个例子

System.out.println(
    Stream.generate(new AtomicInteger()::incrementAndGet)
        .parallel()
        .map(String::valueOf)
        .limit(10)
        .collect(Collectors.toList())
);

在我的机器上打印类似的内容（JDK-14）

[4, 8, 5, 6, 10, 3, 7, 1, 9, 11]

使用 JDK-8，它甚至会打印类似的内容

[4, 14, 18, 24, 30, 37, 42, 52, 59, 66]

如果一个结构像

---

IntStream.range(0, n).mapToObj(i -> myReader.read())

由于未使用

i

参数，感觉很奇怪，你可以使用

Collections.nCopies(n, myReader).stream().map(TypeOfMyReader::read)

相反。这不会显示未使用的 

int

参数，并且效果同样好，因为事实上，它在内部实现为

IntStream.range(0, n).mapToObj(i -> element)

。没有办法绕过某些计数器（无论是可见的还是隐藏的）来确保该方法将被调用

n

次。请注意，由于 read 可能是有状态操作，因此在启用并行处理时，结果行为将始终类似于无序流，但

IntStream

和

nCopies

方法创建了一个有限流，该流永远不会调用该方法超过指定次数。

Answer 2

class MyStreamSpliterator implements Spliterator<String> { // or whichever datatype
    private final MyReaderClass reader;

    public MyStramSpliterator(MyReaderClass reader) {
        this.reader = reader;
    }

    @Override
    public boolean tryAdvance(Consumer<String> action) {
        try {
            String nextval = reader.read();
            action.accept(nextval);
            return true;
        } catch(NoSuchElementException e) {
            // cleanup if necessary
            return false;
        }
        // Alternative: if you really really want to use n iterations,
        // add a counter and use it.
    }

    @Override
    public Spliterator<String> trySplit() {
        return null; // we don't split
    }

    @Override
    public long estimateSize() {
        return Long.MAX_VALUE; // or the correct value, if you know it before
    }

    @Override
    public int characteristics() {
        // add SIZED if you know the size
        return Spliterator.IMMUTABLE | Spliterator.ORDERED;
    }
}

然后，将您的流创建为 

StreamSupport.stream(new MyStreamSpliterator(reader), false)

免责声明：我只是将其放在SO编辑器中，可能存在一些错误。

Answer 3

Collections.nCopies

 与明显的虚拟值一起使用，可以更明显地看出初始流值未使用，并且目标是准确调用该方法 

n 次：

Stream<String> myStream =
        Collections.nCopies(n, null).stream().map(_ -> myReader.read());

类似地，可以使用具有限制的虚拟生成器函数，然后将所得固定大小流的每个虚拟元素映射到

read

值的结果：

Stream<String> myStream =
        Stream.generate(() -> null).limit(n).map(_ -> myReader.read());