为什么Java Streams一次性关闭？

Question

与C#不同IEnumerable,执行管道可以根据需要执行多次,在Java中,流只能"迭代"一次.

对终端操作的任何调用都会关闭流,使其无法使用.这个"功能"消耗了很多力量.

我想这个的原因不是技术性的.这个奇怪限制背后的设计考虑是什么？

编辑:为了演示我在说什么,请考虑以下C#中的Quick-Sort实现:

IEnumerable<int> QuickSort(IEnumerable<int> ints)
{
  if (!ints.Any()) {
    return Enumerable.Empty<int>();
  }

  int pivot = ints.First();

  IEnumerable<int> lt = ints.Where(i => i < pivot);
  IEnumerable<int> gt = ints.Where(i => i > pivot);

  return QuickSort(lt).Concat(new int[] { pivot }).Concat(QuickSort(gt));
}

现在可以肯定的是,我并不是说这是一个很好的快速排序!然而,它是lambda表达式与流操作相结合的表达能力的一个很好的例子.

它不能用Java完成!我甚至无法询问流是否为空而不使其无法使用.

Answer 1

我对Streams API的早期设计有一些回忆,可能会对设计原理有所了解.

早在2012年,我们就在语言中添加了lambda,我们想要一个面向集合或"批量数据"的操作集,使用lambdas编程,这将有助于并行性.在这一点上,懒洋洋地将操作链接在一起的想法已经确立.我们也不希望中间操作存储结果.

我们需要确定的主要问题是链中的对象在API中的样子以及它们如何连接到数据源.源通常是集合,但我们也希望支持来自文件或网络的数据,或者即时生成的数据,例如来自随机数生成器的数据.

现有工作对设计有很多影响.其中影响力最大的是谷歌的Guava图书馆和Scala馆藏图书馆.(如果有人对Guava的影响感到惊讶,请注意,Guava首席开发人员Kevin Bourrillion是JSR-335 Lambda专家组的成员.)在Scala系列中,我们发现Martin Odersky的这篇演讲特别引人关注:Future- Scala集合的证明:从可变到持久到并行.(斯坦福大学EE380,2011年6月1日)

我们当时的原型设计基于Iterable.熟悉的操作filter,map等等是扩展(默认)方法Iterable.调用一个操作添加了一个操作链并返回另一个操作Iterable.终端操作就像count将iterator()链调用到源,并且操作在每个阶段的迭代器中实现.

由于这些是Iterables,您可以iterator()多次调用该方法.那会发生什么？

如果源是一个集合,这大多数工作正常.集合是可iterator()迭代的,每次调用都会生成一个独立于任何其他活动实例的独特Iterator实例,并且每个实例都独立遍历集合.大.

现在,如果源是一次性的,如从文件中读取行,该怎么办？也许第一个迭代器应该获得所有值,但第二个和后续迭代器应该是空的.也许值应该在迭代器之间交错.或者也许每个迭代器都应该得到所有相同的值.那么,如果你有两个迭代器而另一个比另一个更远呢？有人必须缓冲第二个迭代器中的值,直到它们被读取为止.更糟糕的是,如果你得到一个迭代器并读取所有值,然后再获得第二个迭代器,该怎么办？这些价值从何而来？有没有为他们所有的要求来进行缓冲起来以防万一有人想第二个迭代器？

显然,在一次性源上允许多个迭代器会引发很多问题.我们没有他们的好答案.如果你打iterator()两次电话,我们想要一致,可预测的行为.这促使我们不允许多次遍历,使管道一次性完成.

我们还观察到其他人遇到了这些问题.在JDK中,大多数Iterables是集合或类似集合的对象,它们允许多次遍历.它没有在任何地方指定,但似乎有一个不成文的期望Iterables允许多次遍历.一个值得注意的例外是NIO DirectoryStream接口.它的规范包括这个有趣的警告:

虽然DirectoryStream扩展了Iterable,但它不是通用的Iterable,因为它只支持一个Iterator; 调用iterator方法来获取第二个或后续的迭代器会抛出IllegalStateException.

[原件大胆]

这看起来很不寻常和不愉快,我们不想创建一堆可能只有一次的新Iterables.这促使我们远离使用Iterable.

大约在这个时候,布鲁斯·埃克尔(Bruce Eckel)发表的一篇文章描述了他与斯卡拉(Scala)一起遇到的麻烦.他写了这段代码:

// Scala
val lines = fromString(data).getLines
val registrants = lines.map(Registrant)
registrants.foreach(println)
registrants.foreach(println)

这很简单.它将文本行解析为Registrant对象并将其打印两次.除了它实际上只打印出一次.事实证明,他认为这registrants是一个集合,实际上它是一个迭代器.第二次调用foreach遇到一个空迭代器,所有值都已耗尽,所以它什么都不打印.

这种经验使我们相信,如果尝试多次遍历,那么获得明确可预测的结果非常重要.它还强调了区分类似管道的惰性结构与存储数据的实际集合的重要性.这反过来将延迟管道操作分离到新的Stream接口,并且只在集合上直接保留急切的变异操作.Brian Goetz解释了这个理由.

如何允许多次遍历基于集合的管道,但不允许基于非集合的管道？这是不一致的,但这是明智的.如果您正在从网络中读取值,当然您无法再次遍历它们.如果要多次遍历它们,则必须明确地将它们拖入集合中.

但是让我们探讨允许从基于集合的管道进行多次遍历.假设你这样做了:

Iterable<?> it = source.filter(...).map(...).filter(...).map(...);
it.into(dest1);
it.into(dest2);

(into现在拼写了这个操作collect(toList()).)

如果source是一个集合,那么第一次into()调用将创建一个迭代链返回源,执行管道操作,并将结果发送到目标.第二次调用into()将创建另一个迭代器链,并再次执行管道操作.这显然不是错误的,但它确实具有为每个元素第二次执行所有过滤器和映射操作的效果.我想很多程序员会对这种行为感到惊讶.

正如我上面提到的,我们一直在与Guava开发人员交谈.他们有一个很酷的东西是Idea Graveyard,他们描述了他们决定不实施的功能以及原因.懒惰收藏的想法听起来很酷,但这是他们对此所说的.考虑一个List.filter()返回以下内容的操作List:

这里最大的担忧是太多的操作成为昂贵的线性时间命题.如果你想过滤一个列表并获得一个列表,而不仅仅是一个Collection或Iterable,你可以使用ImmutableList.copyOf(Iterables.filter(list, predicate)),它" 预先说明"它正在做什么以及它有多贵.

举一个具体的例子,什么是成本get(0)或size()上的列表？对于常用的类ArrayList,它们是O(1).但是如果你在一个延迟过滤的列表中调用其中一个,它必须在支持列表上运行过滤器,并且突然这些操作是O(n).更糟糕的是,它必须遍历每个操作的后备列表.

在我们看来,这太懒了.设置一些操作并推迟实际执行是一回事,直到你"Go"为止.另一种方法是以一种隐藏潜在大量重新计算的方式进行设置.

在提议禁止非线性或"不再使用"流时,Paul Sandoz描述了允许它们产生"意外或混乱结果" 的潜在后果.他还提到并行执行会使事情更棘手.最后,我补充说,如果操作意外地执行了多次,或者至少与程序员预期的次数不同,那么带有副作用的管道操作会导致困难和模糊的错误.(但Java程序员不会编写带副作用的lambda表达式,是吗？做他们吗??)

因此,这是Java 8 Streams API设计的基本原理,允许一次性遍历,并且需要严格的线性(无分支)管道.它提供跨多个不同流源的一致行为,它清楚地将懒惰与急切操作分开,并且它提供了一种简单的执行模型.

关于IEnumerable,我远不是C#和.NET的专家,所以如果我得出任何不正确的结论,我将不胜感激(温和地).但是,它似乎IEnumerable允许多次遍历在不同的源上表现不同; 并且它允许嵌套IEnumerable操作的分支结构,这可能导致一些重要的重新计算.虽然我理解不同的系统做出不同的权衡,但这些是我们在设计Java 8 Streams API时要避免的两个特征.

OP给出的快速举例很有意思,令人费解,我很遗憾地说,有点可怕.调用QuickSort取一个IEnumerable并返回一个IEnumerable,所以在IEnumerable遍历最后一个之前不会实际排序.但是,调用似乎要做的是构建一个树结构,IEnumerables它反映了quicksort可以执行的分区,而不是实际执行它.(毕竟这是懒惰的计算.)如果源有N个元素,那么树的最宽处是N个元素,并且它将是lg(N)级别的深度.

在我看来 - 再次,我不是C#或.NET专家 - 这将导致某些看似无害的调用,例如透视选择ints.First(),比他们看起来更昂贵.当然,在第一级,它是O(1).但是考虑在树的深处,在右边缘.要计算此分区的第一个元素,必须遍历整个源,即O(N)操作.但由于上面的分区是惰性的,因此必须重新计算它们,需要进行O(lg N)比较.因此,选择枢轴将是O(N lg N)操作,这与整个排序一样昂贵.

但是我们实际上并没有排序,直到我们遍历返回IEnumerable.在标准快速排序算法中,每个分区级别使分区数量加倍.每个分区只有一半大小,因此每个级别都保持O(N)复杂度.分区树的高度为O(lg N),因此总工作量为O(N lg N).

使用懒惰的IEnumerables树,在树的底部有N个分区.计算每个分区需要遍历N个元素,每个元素都需要在树上进行lg(N)比较.要计算树底部的所有分区,则需要进行O(N ^ 2 lg N)比较.

(这是对的吗？我简直不敢相信.有人请为我检查一下.)

无论如何,IEnumerable以这种方式构建复杂的计算结构确实很酷.但是,如果它确实增加了我认为的计算复杂度,那么这种方式的编程似乎是应该避免的,除非一个人非常小心.

Answer 2

背景

虽然问题看似简单,但实际答案需要一些背景才有意义.如果您想跳到结论,请向下滚动...

选择您的比较点 - 基本功能

使用基本概念,C#的IEnumerable概念与JavaIterable密切相关,Java可以根据需要创建尽可能多的迭代器.IEnumerables创造IEnumerators.Java的Iterable创造Iterators

每个概念的历史是相似的,在这两个IEnumerable和Iterable有一个基本的动机,让"换每个"风格遍历数据收集的成员.这是一个过于简单化,因为他们不仅仅允许这一点,他们也通过不同的进展到达那个阶段,但它是一个重要的共同特征,无论如何.

让我们比较一下这个特性:在两种语言中,如果一个类实现了IEnumerable/ Iterable,那么该类必须至少实现一个方法(对于C#,GetEnumerator对于Java而言,它是iterator()).在每种情况下,从(IEnumerator/ Iterator)返回的实例允许您访问数据的当前和后续成员.此功能用于for-each语言语法.

选择您的比较点 - 增强功能

IEnumerable在C#中已经扩展到允许许多其他语言功能(主要与Linq相关).添加的功能包括选择,投影,聚合等.这些扩展具有强烈的动机,从集合理论中使用,类似于SQL和关系数据库概念.

Java 8还添加了一些功能,可以使用Streams和Lambdas进行一定程度的函数编程.请注意,Java 8流不是主要由集合论推动,而是通过函数式编程.无论如何,有很多相似之处.

所以,这是第二点.对C#的增强实现了对IEnumerable概念的增强.在Java中,虽然所做的改进是通过创建lambda表达式和流的新基地的概念,然后又创造了比较琐碎的方式从转换实施Iterators和Iterables溪流,和反之亦然.

因此,将IEnumerable与Java的Stream概念进行比较是不完整的.您需要将它与Java中的组合Streams和Collections API进行比较.

在Java中,Streams与Iterables或Iterators不同

流不是以迭代器的方式解决问题的:

• 迭代器是一种描述数据序列的方法.
• 流是描述数据转换序列的一种方式.

使用a Iterator,您将获得数据值,对其进行处理,然后获取另一个数据值.

使用Streams,您可以将一系列函数链接在一起,然后将输入值提供给流,并从组合序列中获取输出值.注意,在Java术语中,每个函数都封装在一个Stream实例中.Streams API允许您以链接Stream一系列转换表达式的方式链接一系列实例.

为了完成这个Stream概念,您需要一个数据源来提供流,以及一个消耗流的终端功能.

将值输入到流中的方式实际上可能来自a Iterable,但Stream序列本身不是a Iterable,它是复合函数.

A Stream也是懒惰的,因为只有当你从它请求一个值时它才起作用.

请注意Streams的这些重要假设和功能:

• 一个Stream在Java是一种转换引擎,它把一个数据项的一个状态,到另一个状态.
• 流没有数据顺序或位置的概念,简单地转换它们被要求的任何东西.
• 流可以提供来自许多来源的数据,包括其他流,迭代器,Iterables,集合,
• 你不能"重置"一个流,就像"重新编程转换".重置数据源可能就是您想要的.
• 流中任何时候逻辑上只有1个数据项"在飞行中"(除非流是并行流,此时每个线程有1个项).这与数据源无关,该数据源可能具有多于当前项"准备好"提供给流,或者流收集器可能需要聚合和减少多个值.
• 流可以是未绑定的(无限的),仅受数据源或收集器的限制(也可以是无限的).
• 流是"可链接的",过滤一个流的输出是另一个流.输入到流并由流转换的值又可以提供给执行不同转换的另一个流.处于转换状态的数据从一个流流向下一个流.您无需干预并从一个流中提取数据并将其插入下一个流中.

C#比较

当您认为Java Stream只是供应,流和收集系统的一部分,并且Streams和Iterators经常与Collections一起使用时,难怪很难与相同的概念相关联.几乎所有嵌入到IEnumerableC#中的单个概念.

IEnumerable(和密切相关的概念)的部分在所有Java Iterator,Iterable,Lambda和Stream概念中都很明显.

Java概念可以做的小事情在IEnumerable中更难,反之亦然.

结论

• 这里没有设计问题,只是语言之间匹配概念的问题.
• Streams以不同的方式解决问题
• Streams为Java添加了功能(它们添加了不同的处理方式,它们不会占用功能)

添加流可以在解决问题时为您提供更多选择,这可以归类为"增强能力",而不是"减少","带走"或"限制"它.

为什么Java Streams一次性关闭？

这个问题是错误的,因为流是功能序列,而不是数据.根据提供流的数据源,您可以重置数据源,并提供相同或不同的流.

与C#的IEnumerable不同,执行管道可以根据需要执行多次,在Java中,流只能"迭代"一次.

比较a IEnumerable与a Stream是误导的.您使用的上下文IEnumerable可以根据需要执行多次,最好与Java相比,Java Iterables可以根据需要进行多次迭代.Java Stream表示IEnumerable概念的子集,而不是提供数据的子集,因此不能"重新运行".

对终端操作的任何调用都会关闭流,使其无法使用.这个"功能"消耗了很多力量.

从某种意义上说,第一种说法是正确的."夺走权力"声明不是.你还在比较它的IEnumerables.流中的终端操作就像for循环中的'break'子句.如果需要,您可以随时拥有另一个流,并且可以重新提供所需的数据.同样,如果你认为IEnumerable它更像是一个Iterable,对于这个语句,Java做得很好.

我想这个的原因不是技术性的.这个奇怪限制背后的设计考虑是什么？

原因是技术性的,原因很简单,Stream是认为它的一部分.流子集不控制数据供应,因此您应该重置供应,而不是流.在这种情况下,它并不那么奇怪.

QuickSort示例

您的快速排序示例具有签名:

IEnumerable<int> QuickSort(IEnumerable<int> ints)

您将输入IEnumerable视为数据源:

IEnumerable<int> lt = ints.Where(i => i < pivot);

此外,返回值也是IEnumerable一个数据供应,由于这是一个排序操作,因此该供应的顺序很重要.如果你考虑的Java Iterable类是这个合适的匹配,特别是List专业化Iterable的,因为名单是具有保证的顺序或迭代数据的提供,则相应的Java代码,你的代码是:

Stream<Integer> quickSort(List<Integer> ints) {
    // Using a stream to access the data, instead of the simpler ints.isEmpty()
    if (!ints.stream().findAny().isPresent()) {
        return Stream.of();
    }

    // treating the ints as a data collection, just like the C#
    final Integer pivot = ints.get(0);

    // Using streams to get the two partitions
    List<Integer> lt = ints.stream().filter(i -> i < pivot).collect(Collectors.toList());
    List<Integer> gt = ints.stream().filter(i -> i > pivot).collect(Collectors.toList());

    return Stream.concat(Stream.concat(quickSort(lt), Stream.of(pivot)),quickSort(gt));
}    

请注意,有一个错误(我已经复制了),因为排序不能优雅地处理重复值,它是一种"唯一值"排序.

还要注意Java代码如何使用数据源(List),以及在不同点流式传输概念,而在C#中,这两个"个性"可以用于表达IEnumerable.此外,虽然我已经使用List了基本类型,但我可以使用更通用的Collection,并且通过一个小的迭代器到流转换,我可以使用更一般的Iterable

Answer 3

Streams是围绕Spliterators 建立的,它们是有状态的,可变的对象.他们没有"重置"动作,事实上,要求支持这种倒带动作会"带走很多力量".怎么会Random.ints()处理这样的请求？

另一方面,对于Stream具有可回溯原点的s,很容易构造相当于Stream再次使用的等价物.只需将构建的步骤构建Stream为可重用的方法.请记住,重复这些步骤并不是一项昂贵的操作,因为所有这些步骤都是惰性操作; 实际工作从终端操作开始,根据实际的终端操作,可能会执行完全不同的代码.

作为这种方法的作者,你需要指定调用该方法两次暗示的内容:它是否重现完全相同的序列,如为未修改的数组或集合创建的流所做的那样,或者它是否产生具有类似的语义,但不同的元素,如随机int流或控制台输入行流等.

顺便说一下,为了避免混淆,终端操作消耗的Stream是从不同的闭合的Stream作为调用close()流上做(这是需要的具有相关联的资源,如通过产生,例如流Files.lines()).

这似乎很混乱,从误导的比较茎IEnumerable用Stream.An IEnumerable表示提供实际的能力IEnumerator,因此它类似于IterableJava.相比之下,a Stream是一种迭代器并且可以IEnumerator与之相媲美,因此声称这种数据类型可以在.NET中多次使用是错误的,支持IEnumerator.Reset是可选的.这里讨论的例子更多地使用了一个事实,即一个IEnumerable可以用来获取新 IEnumerator的,并且也适用于Java的Collection; 你可以得到一个新的Stream.如果Java开发人员决定直接添加Stream操作Iterable,中间操作返回另一个操作Iterable,那么它实际上是可比较的,它可以以相同的方式工作.

但是,开发人员决定反对它,并在这个问题中讨论了这个决定.最重要的一点是关于渴望收集操作和懒惰流操作的混淆.通过查看.NET API,我(是的,亲自)认为它是合理的.虽然看起来很合理IEnumerable,但是一个特定的Collection会有很多方法直接操作Collection,而且很多方法都会返回一个懒惰的方法IEnumerable,而方法的特殊性并不总是直观可识别的.我发现的最糟糕的例子(在我查看它的几分钟内)List.Reverse()的名称与继承的名称完全匹配(这是扩展方法的正确终点吗？),Enumerable.Reverse()同时具有完全矛盾的行为.

当然,这是两个截然不同的决定.第一个使Stream类型与Iterable/ 不同,Collection第二个Stream使用一种迭代器而不是另一种迭代.但是这些决定是在一起进行的,可能是因为分离这两个决定从未被考虑过.它不是为了与.NET的可比性而创建的.

实际的API设计决定是添加一种改进的迭代器类型Spliterator.Spliterators可以由旧的Iterables(这是改装的方式)或全新的实现提供.然后,Stream作为高级前端添加到相当低级别的Spliterators.而已.您可以讨论一个不同的设计是否会更好,但这不是很有效率,它不会改变,因为它们现在的设计方式.

您还需要考虑另一个实施方面.Streams 不是不可变的数据结构.每个中间操作可以返回一个Stream封装旧实例的新实例,但它也可以操纵它自己的实例并返回它自己(这也不排除对同一个操作都做同样的操作).通常已知的示例是类似parallel或unordered不添加另一步但操纵整个管道的操作.拥有这样一个可变数据结构并尝试重用(或者更糟糕的是,同时多次使用它)并不能很好地发挥作用......

为了完整起见,这里将您的quicksort示例翻译为Java StreamAPI.它表明它并没有真正"带走太多力量".

static Stream<Integer> quickSort(Supplier<Stream<Integer>> ints) {

  final Optional<Integer> optPivot = ints.get().findAny();
  if(!optPivot.isPresent()) return Stream.empty();

  final int pivot = optPivot.get();

  Supplier<Stream<Integer>> lt = ()->ints.get().filter(i -> i < pivot);
  Supplier<Stream<Integer>> gt = ()->ints.get().filter(i -> i > pivot);

  return Stream.of(quickSort(lt), Stream.of(pivot), quickSort(gt)).flatMap(s->s);
}

它可以像

List<Integer> l=new Random().ints(100, 0, 1000).boxed().collect(Collectors.toList());
System.out.println(l);
System.out.println(quickSort(l::stream)
    .map(Object::toString).collect(Collectors.joining(", ")));

你可以写得更紧凑

static Stream<Integer> quickSort(Supplier<Stream<Integer>> ints) {
    return ints.get().findAny().map(pivot ->
         Stream.of(
                   quickSort(()->ints.get().filter(i -> i < pivot)),
                   Stream.of(pivot),
                   quickSort(()->ints.get().filter(i -> i > pivot)))
        .flatMap(s->s)).orElse(Stream.empty());
}

Answer 4

我认为,当你仔细观察时,两者之间的差异很小.

在它的脸上,IEnumerable似乎确实是一个可重用的构造:

IEnumerable<int> numbers = new int[] { 1, 2, 3, 4, 5 };

foreach (var n in numbers) {
    Console.WriteLine(n);
}

但是,编译器实际上正在做一些工作来帮助我们; 它生成以下代码:

IEnumerable<int> numbers = new int[] { 1, 2, 3, 4, 5 };

IEnumerator<int> enumerator = numbers.GetEnumerator();
while (enumerator.MoveNext()) {
    Console.WriteLine(enumerator.Current);
}

每次实际迭代可枚举时,编译器都会创建一个枚举器.普查员不可重复使用; 进一步调用MoveNext将返回false,并且无法将其重置为开头.如果要再次迭代数字,则需要创建另一个枚举器实例.

为了更好地说明IEnumerable具有(可以具有)与Java Stream相同的"功能",请考虑其数字源不是静态集合的枚举.例如,我们可以创建一个可枚举的对象,该对象生成一个包含5个随机数的序列:

class Generator : IEnumerator<int> {
    Random _r;
    int _current;
    int _count = 0;

    public Generator(Random r) {
        _r = r;
    }

    public bool MoveNext() {
        _current= _r.Next();
        _count++;
        return _count <= 5;
    }

    public int Current {
        get { return _current; }
    }
 }

class RandomNumberStream : IEnumerable<int> {
    Random _r = new Random();
    public IEnumerator<int> GetEnumerator() {
        return new Generator(_r);
    }
    public IEnumerator IEnumerable.GetEnumerator() {
        return this.GetEnumerator();
    }
}

现在我们的代码与之前基于数组的可枚举代码非常相似,但是第二次迭代结束numbers:

IEnumerable<int> numbers = new RandomNumberStream();

foreach (var n in numbers) {
    Console.WriteLine(n);
}
foreach (var n in numbers) {
    Console.WriteLine(n);
}

我们第二次迭代,numbers我们将得到一个不同的数字序列,这在同一意义上是不可重用的.或者,RandomNumberStream如果您尝试多次迭代它,我们可以写入抛出异常,使可枚举实际上无法使用(如Java Stream).

另外,基于可枚举的快速排序在应用于RandomNumberStream什么时意味着什么？

结论

因此,最大的区别在于,.NET允许您在需要访问序列中的元素时IEnumerable通过隐式IEnumerator在后台创建新内容来重用.

这种隐式行为通常很有用(并且在你陈述时是"强大的"),因为我们可以反复迭代一个集合.

但有时,这种隐性行为实际上可能会导致问题.如果您的数据源不是静态的,或者访问成本很高(如数据库或网站),那么IEnumerable必须抛弃许多假设; 重用不是那么直截了当