Java 8流串行与并行性能

Question

在我的机器上，以下程序打印：

OptionalLong[134043]
 PARALLEL took 127869 ms
OptionalLong[134043]
 SERIAL took 60594 ms

我不清楚为什么串行执行程序要比并行执行程序快。我把这两个程序-Xms2g -Xmx2g放在8gb相对安静的盒子上。有人可以澄清发生了什么吗？

import java.util.stream.LongStream;
import java.util.stream.LongStream.Builder;

public class Problem47 {

    public static void main(String[] args) {

        final long startTime = System.currentTimeMillis();
        System.out.println(LongStream.iterate(1, n -> n + 1).parallel().limit(1000000).filter(n -> fourConsecutives(n)).findFirst());
        final long endTime = System.currentTimeMillis();
        System.out.println(" PARALLEL took " +(endTime - startTime) + " ms");

        final long startTime2 = System.currentTimeMillis();
        System.out.println(LongStream.iterate(1, n -> n + 1).limit(1000000).filter(n -> fourConsecutives(n)).findFirst());
        final long endTime2 = System.currentTimeMillis();
        System.out.println(" SERIAL took " +(endTime2 - startTime2) + " ms");
    }

    static boolean fourConsecutives(final long n) {
        return distinctPrimeFactors(n).count() == 4 &&
                distinctPrimeFactors(n + 1).count() == 4 &&
                distinctPrimeFactors(n + 2).count() == 4 &&
                distinctPrimeFactors(n + 3).count() == 4;
    }

    static LongStream distinctPrimeFactors(long number) {
        final Builder builder = LongStream.builder();
        final long limit = number / 2;
        long n = number;
        for (long i = 2; i <= limit; i++) {
            while (n % i == 0) {
                builder.accept(i);
                n /= i;
            }
        }
        return builder.build().distinct();
    }

}

Answer 1

尽管Brian Goetz对您的设置是正确的，例如，您应该使用.range(1, 1000000)而不是.iterate(1, n -> n + 1).limit(1000000)，并且您的基准测试方法非常简单，但我想强调以下要点：

即使解决了这些问题，甚至使用挂钟和TaskManager也可以看到有问题。在我的机器上，该操作大约需要半分钟，并且您可以看到大约两秒钟后并行性下降到了单核。即使专业的基准测试工具可以产生不同的结果，也没关系，除非您想一直在基准测试工具中运行最终应用程序……

现在我们可以尝试更多地模拟您的设置，或者告诉您应该像在讨论列表中的实现者那样学习有关Fork / Join框架的特殊知识。

或者我们尝试另一种实现方式：

ExecutorService es=Executors.newFixedThreadPool(
                       Runtime.getRuntime().availableProcessors());
AtomicLong found=new AtomicLong(Long.MAX_VALUE);
LongStream.range(1, 1000000).filter(n -> found.get()==Long.MAX_VALUE)
    .forEach(n -> es.submit(()->{
        if(found.get()>n && fourConsecutives(n)) for(;;) {
            long x=found.get();
            if(x<n || found.compareAndSet(x, n)) break;
        }
    }));
es.shutdown();
try { es.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS); }
catch (InterruptedException ex) {throw new AssertionError(ex); }
long result=found.get();
System.out.println(result==Long.MAX_VALUE? "not found": result);

在我的机器上，它完成了我期望的并行执行所需要的时间仅多于⟨sequential time⟩/⟨number of cpu cores⟩。无需更改您的fourConsecutives实现。

底线是，至少当处理单个项目需要花费大量时间时，当前的Stream实现（或底层的Fork /
Join框架）具有如在此相关问题中已经讨论的问题。如果您想要可靠的并行性，我建议使用经过验证和测试ExecutorService的。在我的示例中可以看到，这并不意味着放弃Java
8功能，它们很好地结合在一起。仅考虑引入的自动并行性Stream.parallel（考虑到当前的实现）。