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

Question

我想复制一个Java 8流，以便可以处理两次。我可以collect列出并从中获得新的信息流；// doSomething() returns a streamList<A> thing = doSomething().collect(toList());thing.stream()... // do stuffthing.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) {

    final List<T> gap = new LinkedList<>();

    final Iterator<T> it = stream.iterator();

    @SuppressWarnings("unchecked")

    final Iterator<T>[] ahead = new Iterator[] { null };

    class Duplicate implements Iterator<T> {

        @Override

        public boolean hasNext() {

            if (ahead[0] == null || ahead[0] == this)

                return it.hasNext();

            return !gap.isEmpty();

        }

        @Override

        public T next() {

            if (ahead[0] == null)

                ahead[0] = this;

            if (ahead[0] == this) {

                T value = it.next();

                gap.offer(value);

                return value;

            }

            return gap.poll();

        }

    }

    return tuple(seq(new Duplicate()), seq(new Duplicate()));

}

更多源代码在这里

Tuple2大概是喜欢你的Pair类型，而Seq为Stream一些增强功能。