使用大型并行 Java 8 流时如何防止堆空间错误

Question

如何有效地并行计算 pi（仅作为示例）？

这可行（在我的机器上大约需要 15 秒）：

Stream.iterate(1d, d->-(d+2*(Math.abs(d)/d))).limit(999999999L).mapToDouble(d->4.0d/d).sum()

但以下所有并行变体都会遇到 OutOfMemoryError

DoubleStream.iterate(1d, d->-(d+2*(Math.abs(d)/d))).parallel().limit(999999999L).map(d->4.0d/d).sum();
DoubleStream.iterate(1d, d->-(d+2*(Math.abs(d)/d))).limit(999999999L).parallel().map(d->4.0d/d).sum();
DoubleStream.iterate(1d, d->-(d+2*(Math.abs(d)/d))).limit(999999999L).map(d->4.0d/d).parallel().sum();

那么，我需要做什么才能并行处理这个（大）流？ 我已经检查过自动装箱是否导致内存消耗，但事实并非如此。这也有效：

DoubleStream.iterate(1, d->-(d+Math.abs(2*d)/d)).boxed().limit(999999999L).mapToDouble(d->4/d).sum()

Answer 1

问题在于您使用的结构难以并行化。

首先，Stream.iterate(…) 创建了一个数字序列，其中每个计算都依赖于前一个值，因此，它没有为并行计算提供空间。更糟糕的是，它创建了一个无限流，将​​由实现处理，就像一个未知大小的流。为了拆分流，必须先将值收集到数组中，然后才能将它们交给其他计算线程。

其次，提供limit(…) 并不能改善情况，it makes the situation even worse。应用限制会删除实现刚刚为数组片段收集的大小信息。原因是流是有序的，因此处理数组片段的线程不知道它是否可以处理所有元素，因为这取决于其他线程正在处理多少先前元素的信息。这是documented：

“...在有序的并行管道上可能会非常昂贵，尤其是对于 maxSize 的大值，因为 limit(n) 不仅返回任何 n 元素，而且返回 相遇顺序中的前 n 个元素。”

很遗憾，因为我们完全知道iterate 返回的无限序列与limit(…) 的组合实际上具有完全已知的大小。但是实现不知道。而且 API 没有提供一种方法来创建两者的有效组合。但我们可以自己做：

static DoubleStream iterate(double seed, DoubleUnaryOperator f, long limit) {
  return StreamSupport.doubleStream(new Spliterators.AbstractDoubleSpliterator(limit,
     Spliterator.ORDERED|Spliterator.SIZED|Spliterator.IMMUTABLE|Spliterator.NONNULL) {
       long remaining=limit;
       double value=seed;
       public boolean tryAdvance(DoubleConsumer action) {
           if(remaining==0) return false;
           double d=value;
           if(--remaining>0) value=f.applyAsDouble(d);
           action.accept(d);
           return true;
       }
   }, false);
}

一旦我们有了这样一个 iterate-with-limit 方法，我们就可以像这样使用它

iterate(1d, d -> -(d+2*(Math.abs(d)/d)), 999999999L).parallel().map(d->4.0d/d).sum()

由于源代码的顺序性，这仍然不能从并行执行中获得太多好处，但它确实有效。在我的四核机器上，它设法获得了大约 20% 的增益。

Answer 2

这是因为parallel() 方法使用的默认ForkJoinPool 实现不限制创建的线程数。解决方案是提供ForkJoinPool 的自定义实现，该实现受限于它并行执行的线程数。这可以通过如下方式实现：

ForkJoinPool forkJoinPool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());
forkJoinPool.submit(() -> DoubleStream.iterate(1d, d->-(d+2*(Math.abs(d)/d))).parallel().limit(999999999L).map(d->4.0d/d).sum());