Java 8，在流中使用.parallel会导致OOM错误

Question

在 Java 8 In Action
一书的第7.1.1节中，作者指出，通过添加功能，流可以从并行处理中受益.parallel()。他们提供了一个简单的方法parallelSum(int)来说明这一点。我很好奇它的工作原理，所以我执行了以下代码：

package lambdasinaction.chap7;

import java.util.stream.Stream;

public class ParallelPlay {

    public static void main(String[] args) {
        System.out.println(parallelSum(100_000_000));
    }

    public static long parallelSum(long n) {
        return Stream.iterate(1L, i -> i + 1)
                .limit(n)
                .parallel()
                .reduce(0L, Long::sum);
    }
}

令我惊讶的是，我收到此错误：

Exception in thread "main" java.lang.OutOfMemoryError
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(Unknown Source)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(Unknown Source)
    at java.lang.reflect.Constructor.newInstance(Unknown Source)
    at java.util.concurrent.ForkJoinTask.getThrowableException(Unknown Source)
    at java.util.concurrent.ForkJoinTask.reportException(Unknown Source)
    at java.util.concurrent.ForkJoinTask.invoke(Unknown Source)
    at java.util.stream.SliceOps$1.opEvaluateParallelLazy(Unknown Source)
    at java.util.stream.AbstractPipeline.sourceSpliterator(Unknown Source)
    at java.util.stream.AbstractPipeline.evaluate(Unknown Source)
    at java.util.stream.ReferencePipeline.reduce(Unknown Source)
    at lambdasinaction.chap7.ParallelPlay.parallelSum(ParallelPlay.java:15)
    at lambdasinaction.chap7.ParallelPlay.main(ParallelPlay.java:8)
Caused by: java.lang.OutOfMemoryError: Java heap space
    at java.util.stream.SpinedBuffer.ensureCapacity(Unknown Source)
    at java.util.stream.Nodes$SpinedNodeBuilder.begin(Unknown Source)
    at java.util.stream.AbstractPipeline.copyInto(Unknown Source)
    at java.util.stream.AbstractPipeline.wrapAndCopyInto(Unknown Source)
    at java.util.stream.SliceOps$SliceTask.doLeaf(Unknown Source)
    at java.util.stream.SliceOps$SliceTask.doLeaf(Unknown Source)
    at java.util.stream.AbstractShortCircuitTask.compute(Unknown Source)
    at java.util.concurrent.CountedCompleter.exec(Unknown Source)
    at java.util.concurrent.ForkJoinTask.doExec(Unknown Source)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(Unknown Source)
    at java.util.concurrent.ForkJoinPool.runWorker(Unknown Source)
    at java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source)

我在Windows 7 SP1上使用四核处理器运行Java 1.8.0_45。这是怎么回事？

Answer 1

在这里，您可以创建无限流并在之后对其进行限制。关于并行处理无限流存在已知问题。特别是，无法有效地将任务分成相等的部分。在内部使用了一些启发式方法，这些启发式方法不太适合每个任务。在您的情况下，最好使用创建有限流LongStream.range：

import java.util.stream.LongStream;

public class ParallelPlay {

    public static void main(String[] args) {
        System.out.println(parallelSum(100_000_000));
    }

    public static long parallelSum(long n) {
        return LongStream.rangeClosed(1, n).parallel().sum();
    }
}

在这种情况下，Stream引擎从一开始就知道您有多少个元素，因此它可以有效地拆分任务。另请注意，使用LongStream会更有效，因为您将没有不必要的装箱。

通常，如果可以用有限的流解决任务，请避免使用无限的流。