复制流以避免"流已被操作或关闭"

Question

我想复制一个Java8流，这样我可以处理它两次。我可以收集作为一个列表，并从中获取新的流；

// doSomething() returns a stream
List<A> thing = doSomething().collect(toList());
thing.stream()... // do stuff
thing.stream()... // do other stuff

但我认为应该有一种更有效/优雅的方式。

有没有办法复制流而不把它变成集合？

我实际上正在处理一个既的流，所以想在移动到右侧投影并以另一种方式处理之前，先以一种方式处理左侧投影。有点像这样（到目前为止，我被迫使用toList技巧）。

List<Either<Pair<A, Throwable>, A>> results = doSomething().collect(toList());

Stream<Pair<A, Throwable>> failures = results.stream().flatMap(either -> either.left());
failures.forEach(failure -> ... );

Stream<A> successes = results.stream().flatMap(either -> either.right());
successes.forEach(success -> ... );




Answer 1

使用供应商为每个终止操作生成流。

Supplier<Stream<Integer>> streamSupplier = () -> list.stream();

每当需要该集合的流时，请使用streamSupplier。get（）以获取新流。

例子：

1. streamSupplier。get（）。anyMatch（谓词）

Answer 2

您可以将局部变量与供应商一起使用，以设置流管道的公共部分。

从…起http://winterbe.com/posts/2014/07/31/java8-stream-tutorial-examples/:

重用流

Java 8流不能被重用。一旦调用任何终端操作，流即关闭：

Stream<String> stream = Stream.of("d2", "a2", "b1", "b3", "c")
    .filter(s -> s.startsWith("a"));
stream.anyMatch(s -> true);    // ok
stream.noneMatch(s -> true);   // exception

Calling `noneMatch` after `anyMatch` on the same stream results in the following exception:
java.lang.IllegalStateException: stream has already been operated upon or closed
at 
java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)
at 
java.util.stream.ReferencePipeline.noneMatch(ReferencePipeline.java:459)
at com.winterbe.java8.Streams5.test7(Streams5.java:38)
at com.winterbe.java8.Streams5.main(Streams5.java:28)

为了克服这一限制，我们必须为我们想要执行的每个终端操作创建一个新的流链，例如，我们可以创建一个流供应商来构建一个新的流，其中所有中间操作都已设置：

Supplier<Stream<String>> streamSupplier =
    () -> Stream.of("d2", "a2", "b1", "b3", "c")
            .filter(s -> s.startsWith("a"));

streamSupplier.get().anyMatch(s -> true);   // ok
streamSupplier.get().noneMatch(s -> true);  // ok

对get（）的每次调用都会构造一个新的流，我们将保存该流以调用所需的终端操作。

Answer 3

我认为你对效率的假设有点倒退。如果你只使用一次数据，你会得到巨大的效率回报，因为你不必存储它，流给你强大的“循环融合”优化，让你在管道中高效地流动整个数据。

如果您想重复使用相同的数据，那么根据定义，您要么生成两次（确定性地）数据，要么存储它。如果它已经在一个集合中，那就太好了；然后迭代两次是很便宜的。

我们用“分叉流”进行了设计实验。我们发现，支持这一点有实际成本；它以牺牲不常见的情况为代价，加重了常见情况（使用一次）的负担。最大的问题是处理“当两个管道不以相同的速度消耗数据时会发生什么”不管怎样，现在你又回到缓冲了。这是一个显然没有分量的功能。

如果要重复操作同一数据，请将其存储，或将操作构造为使用者，然后执行以下操作：

stream()...stuff....forEach(e -> { consumerA(e); consumerB(e); });

您还可以查看RxJava库，因为它的处理模型更适合这种“流分叉”。