并行流处理与线程池处理vs顺序处理

Question

我只是在评估，哪个代码片段在java 8中表现得更好。

代码段1 (在主线程中处理)：

public long doSequence() {
    DoubleStream ds = IntStream.range(0, 100000).asDoubleStream();
    long startTime = System.currentTimeMillis();
    final AtomicLong al = new AtomicLong();
    ds.forEach((num) -> {
        long n1 = new Double (Math.pow(num, 3)).longValue();
        long n2 = new Double (Math.pow(num, 2)).longValue();
        al.addAndGet(n1 + n2);
    });
    System.out.println("Sequence");
    System.out.println(al.get());
    long endTime = System.currentTimeMillis();
    return (endTime - startTime);
}

代码段2 (在并行线程中处理)：

public long doParallel() {
    long startTime = System.currentTimeMillis();
    final AtomicLong al = new AtomicLong();
    DoubleStream ds = IntStream.range(0, 100000).asDoubleStream();
    ds.parallel().forEach((num) -> {
        long n1 = new Double (Math.pow(num, 3)).longValue();
        long n2 = new Double (Math.pow(num, 2)).longValue();
        al.addAndGet(n1 + n2);
    });
    System.out.println("Parallel");
    System.out.println(al.get());
    long endTime = System.currentTimeMillis();
    return (endTime - startTime);
}

代码段3 (从线程池以并行线程方式处理)：

public long doThreadPoolParallel() throws InterruptedException, ExecutionException {
    ForkJoinPool customThreadPool = new ForkJoinPool(4);
    DoubleStream ds = IntStream.range(0, 100000).asDoubleStream();
    long startTime = System.currentTimeMillis();
    final AtomicLong al = new AtomicLong();
    customThreadPool.submit(() -> ds.parallel().forEach((num) -> {
        long n1 = new Double (Math.pow(num, 3)).longValue();
        long n2 = new Double (Math.pow(num, 2)).longValue();
        al.addAndGet(n1 + n2);
    })).get();
    System.out.println("Thread Pool");
    System.out.println(al.get());
    long endTime = System.currentTimeMillis();
    return (endTime - startTime);
}

产出如下：

Parallel
6553089257123798384
34 <--34 milli seconds

Thread Pool
6553089257123798384
23 <--23 milli seconds

Sequence
6553089257123798384
12 <--12 milli seconds!

我期望的是

1)使用线程池进行处理的时间应该是最小的，但不是这样的。(请注意，我没有包括线程池的创建时间，所以应该是快速的)

2)从来没有预料到连续运行的代码会是最快的，原因应该是什么？

我用的是四核处理器。

感谢对解释上述歧义的任何帮助！

Answer 1

您的比较并不完美，当然是因为缺少VM热身。当我简单地重复执行时，我得到了不同的结果：

System.out.println(doParallel());
System.out.println(doThreadPoolParallel());
System.out.println(doSequence());
System.out.println("-------");
System.out.println(doParallel());
System.out.println(doThreadPoolParallel());
System.out.println(doSequence());
System.out.println("-------");
System.out.println(doParallel());
System.out.println(doThreadPoolParallel());
System.out.println(doSequence());

结果：

Parallel
6553089257123798384
65
Thread Pool
6553089257123798384
13
Sequence
6553089257123798384
14
-------
Parallel
6553089257123798384
9
Thread Pool
6553089257123798384
4
Sequence
6553089257123798384
8
-------
Parallel
6553089257123798384
8
Thread Pool
6553089257123798384
3
Sequence
6553089257123798384
8

正如@Erwin在评论中指出的那样，请查看关于这个问题 (本例中的规则1)的答案，以了解如何正确地进行基准测试。

并行流的默认并行性不一定与具有计算机内核的叉连接池提供的并行性相同，尽管当我从自定义池切换到公共分叉连接池时，结果之间的差异仍然可以忽略不计。

Answer 2

AtomicLong.addAndGet需要线程同步--每个线程都必须看到上一个addAndGet的结果--您可以指望总线程是正确的。

虽然这不是传统的synchronized同步，但仍然存在开销。在JDK7中，addAndGet在Java代码中使用了自旋锁。在JDK8中，它被转化为一个内部，然后由HotSpot在Intel平台上发出的一个LOCK:XADD指令来实现。

它需要CPU之间的高速缓存同步，这有一定的开销。它甚至可能需要从主内存中刷新和读取内容，与不需要这样做的代码相比，这是非常慢的。

这是很有可能的，因为这种同步开销发生在测试中的每一次迭代中，这种开销比通过并行所获得的任何性能收益都要大。

参考文献：

• Java中CAS (比较和交换)是如何工作的
• 原子获取和添加vs比较和交换