复制流，以避免“流已被操作或关闭”

Question

我想复制一个Java 8流，以便可以处理两次。我可以collect列出并从中获得新的信息流；

// doSomething() returns a stream

List<A> thing = doSomething().collect(toList());

thing.stream()... // do stuff

thing.stream()... // do other stuff

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

有没有一种方法可以复制流而不将其转换为集合？

我实际上正在使用Eithers 流，因此想先处理左侧投影，然后再移至右侧投影并以另一种方式处理。有点像这样（到目前为止，我被迫使用这种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

我认为您对效率的假设有点倒退。如果您只需要使用一次数据，那么您将获得巨大的效率回报，因为您不必存储数据，而流为您提供了强大的“循环融合”优化，可以使整个数据有效地流经管道。如果您想重复使用相同的数据，那么根据定义，您要么必须生成两次（确定性地），要么将其存储。如果已经在收藏中，那就太好了；然后迭代两次很便宜。我们在“分叉流”中进行了设计实验。我们发现，对此进行支持需要付出实际成本；它负担了普通案例（一次使用）的负担，却以罕见案例为代价。最大的问题是处理“当两个管道不以相同的速率使用数据时会发生什么”。现在您无论如何都要返回缓冲。此功能显然没有发挥作用。如果要重复对相同的数据进行操作，请存储它或将其结构化为“使用者”，然后执行以下操作：stream()...stuff....forEach(e -> { consumerA(e); consumerB(e); });您可能还需要研究RxJava库，因为它的处理模型更适合于这种“流派生”。

Answer 2

我们已经duplicate()在jOOλ中实现了一种流方法，jOOλ是我们创建的一个开放源代码库，用于改进jOOQ的集成测试。本质上，您可以编写：Tuple2<Seq<A>, Seq<A>> duplicates = Seq.seq(doSomething()).duplicate();在内部，有一个缓冲区，用于存储从一个流而不是另一个流消耗的所有值。如果两个流以大约相同的速率消耗，并且如果您可以在没有线程安全的情况下生存，那么这可能与获得的效率一样。该算法的工作原理如下：static <T> Tuple2<Seq<T>, Seq<T>> duplicate(Stream<T> stream) {&nbsp; &nbsp; final List<T> gap = new LinkedList<>();&nbsp; &nbsp; final Iterator<T> it = stream.iterator();&nbsp; &nbsp; @SuppressWarnings("unchecked")&nbsp; &nbsp; final Iterator<T>[] ahead = new Iterator[] { null };&nbsp; &nbsp; class Duplicate implements Iterator<T> {&nbsp; &nbsp; &nbsp; &nbsp; @Override&nbsp; &nbsp; &nbsp; &nbsp; public boolean hasNext() {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if (ahead[0] == null || ahead[0] == this)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return it.hasNext();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return !gap.isEmpty();&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; @Override&nbsp; &nbsp; &nbsp; &nbsp; public T next() {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if (ahead[0] == null)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ahead[0] = this;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if (ahead[0] == this) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; T value = it.next();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; gap.offer(value);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return value;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return gap.poll();&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }&nbsp; &nbsp; return tuple(seq(new Duplicate()), seq(new Duplicate()));}更多源代码在这里Tuple2大概是喜欢你的Pair类型，而Seq为Stream一些增强功能。