并行无限Java流耗尽内存

Question

我试图理解为什么下面的Java程序给出了< code>OutOfMemoryError，而对应的程序却没有< code >。parallel()没有。

System.out.println(Stream
    .iterate(1, i -> i+1)
    .parallel()
    .flatMap(n -> Stream.iterate(n, i -> i+n))
    .mapToInt(Integer::intValue)
    .limit(100_000_000)
    .sum()
);

我有两个问题：

>

该程序的预期输出是什么？

不带<代码>。parallel()看起来这只是输出< code>sum(1 2 3...)这意味着它只是“停留”在平面图中的第一个流，这是有意义的。

对于并行，我不知道是否有预期的行为，但我的猜测是它以某种方式交错了第一个n左右的流，其中n是并行工作人员的数量。根据分块/缓冲行为，它也可能略有不同。

是什么导致它运行内存溢出？我特别想了解这些流是如何在底层实现的。

我猜是有什么东西阻塞了流，所以它永远不会结束，并且能够消除生成的值，但我不太清楚计算的顺序以及缓冲发生的位置。

编辑：如果相关的话，我使用的是Java11。

Editt 2:显然，即使对于简单的程序< code>IntStream.iterate(1，i-

Answer 1

OOME不是因为流是无限的，而是因为它不是无限的。

也就是说，如果你注释掉.limit（...），它永远不会耗尽内存 - 但当然，它也永远不会结束。

一旦分割，如果元素在每个线程中累积，则流只能跟踪元素的数量（看起来实际的累加器是＜code＞Spliterators$ArraySpliterator#array）。

看起来您可以在没有平面图的情况下复制它，只需使用-Xmx128m运行以下命令：

    System.out.println(Stream
            .iterate(1, i -> i + 1)
            .parallel()
      //    .flatMap(n -> Stream.iterate(n, i -> i+n))
            .mapToInt(Integer::intValue)
            .limit(100_000_000)
            .sum()
    );

但是，在注释掉限制（）之后，它应该运行良好，直到您决定备用笔记本电脑为止。

除了实际的实施细节之外，我认为正在发生的事情如下：

使用limit，求和缩减器希望前X个元素求和，因此没有线程可以发出部分和。每个“切片”（线程）都需要累积元素并将其传递。没有限制，没有这样的约束，因此每个“切片”将只计算它得到的元素的部分和（永远），假设它最终将发出结果。

Answer 2

我最好的猜测是，添加parallel（）会改变flatMap（）的内部行为，而flatMap（）之前已经懒洋洋地评估了问题。

[JDK-8202307]在获取java.lang.OutOfMemoryError:java堆空间时调用Stream.iterator（）时报告了您得到的OutOfmemoryErro错误。在平面图中使用无限/非常大的流的流上使用next（）。如果您查看票据，它或多或少与您获得的堆栈跟踪相同。票证已关闭，因为无法修复，原因如下：

迭代器（）和spterator（）方法是“逃逸舱口”，用于无法使用其他操作时使用。它们有一些限制，因为它们将流实现的推送模型转换为拉模型。在某些情况下，这种转换需要缓冲，例如当一个元素被（平面）映射到两个或多个元素时。这将使流实现变得非常复杂，可能是以牺牲常见情况为代价的，以支持背压的概念来传达要通过元素生产的嵌套层提取多少元素。

Answer 3

你说“但是我不太清楚事情是按照什么顺序被评估的，缓冲发生在哪里”，这正是并行流的意义所在。评估顺序未指定。

示例的一个关键方面是限制（100_000_000）。这意味着实现不能只求和任意值，而必须求和前100000000个数字。请注意，在参考实现中，.unordered（）。limit（100_000_000）不会改变结果，这表明对于无序情况没有特殊的实现，但这是一个实现细节。

现在，当工作线程处理元素时，它们不能只是将它们相加，因为它们必须知道允许它们使用哪些元素，这取决于有多少元素在它们的特定工作负载之前。因为这个流不知道大小，这只能在前缀元素被处理后才知道，这对于无限流是永远不会发生的。因此工作线程暂时保持缓冲，这些信息变得可用。

原则上，当工作线程知道它处理最左边的工作块时，它可以立即对元素求和、计数，并在达到极限时发出结束信号。因此流可以终止，但这取决于很多因素。

在您的情况下，一个看似合理的场景是其他工作线程分配缓冲区的速度比最左边的作业计数的速度快。在这种情况下，计时的细微变化可能会使流偶尔返回一个值。

当我们降低所有工作线程(除了处理最左边块的线程)的速度时，我们可以让流终止(至少在大多数运行中):

System.out.println(IntStream
    .iterate(1, i -> i+1)
    .parallel()
    .peek(i -> { if(i != 1) LockSupport.parkNanos(1_000_000_000); })
    .flatMap(n -> IntStream.iterate(n, i -> i+n))
    .limit(100_000_000)
    .sum()
);

我遵循Stuart Marks的建议，在谈论相遇顺序而不是处理顺序时，使用从左到右的顺序。