Java 8的流:为什么并行流更慢？

Question

我正在玩Java 8的流,无法理解我得到的性能结果.我有2个核心CPU(Intel i73520M),Windows 8 x64和64位Java 8更新5.我正在做简单的字符串流/并行流Strings,发现并行版本有点慢.

Function<Stream<String>, Long> timeOperation = (Stream<String> stream) -> {
  long time1 = System.nanoTime();
  final List<String> list = 
     stream
       .map(String::toLowerCase)
       .collect(Collectors.toList());
  long time2 = System.nanoTime();
  return time2 - time1;
};

Consumer<Stream<String>> printTime = stream ->
  System.out.println(timeOperation.apply(stream) / 1000000f);

String[] array = new String[1000000];
Arrays.fill(array, "AbabagalamagA");

printTime.accept(Arrays.stream(array));            // prints around 600
printTime.accept(Arrays.stream(array).parallel()); // prints around 900

考虑到我有2个CPU内核的事实,并行版本不应该更快吗？有人能给我一个暗示为什么并行版本更慢？

Answer 1

这里有几个问题并行发生.

首先,并行解决问题总是涉及执行比按顺序执行更多的实际工作.开销涉及在多个线程之间拆分工作并加入或合并结果.将短字符串转换为小写字符的问题足够小,以至于它们有被并行拆分开销淹没的危险.

第二个问题是Java程序的基准测试是非常微妙的,并且很容易得到令人困惑的结果.两个常见问题是JIT编译和死代码消除.短基准测试通常在JIT编译之前或期间完成,因此它们不会测量峰值吞吐量,实际上它们可能正在测量JIT本身.当编译发生时有些不确定,因此它可能导致结果变化很大.

对于小型综合基准测试,工作负载通常会计算丢弃的结果.JIT编译器非常擅长检测这种情况并消除不会产生任何地方使用结果的代码.在这种情况下可能不会发生这种情况,但如果你修补其他合成工作负载,它肯定会发生.当然,如果JIT消除了基准测试工作负载,它会使基准测试变得无用.

我强烈建议使用一个完善的基准测试框架,如JMH,而不是手动滚动自己的一个.JMH拥有帮助避免常见基准测试陷阱的工具,包括这些陷阱,并且设置和运行起来非常简单.这是您转换为使用JMH的基准:

package com.stackoverflow.questions;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import java.util.concurrent.TimeUnit;

import org.openjdk.jmh.annotations.*;

public class SO23170832 {
    @State(Scope.Benchmark)
    public static class BenchmarkState {
        static String[] array;
        static {
            array = new String[1000000];
            Arrays.fill(array, "AbabagalamagA");
        }
    }

    @GenerateMicroBenchmark
    @OutputTimeUnit(TimeUnit.SECONDS)
    public List<String> sequential(BenchmarkState state) {
        return
            Arrays.stream(state.array)
                  .map(x -> x.toLowerCase())
                  .collect(Collectors.toList());
    }

    @GenerateMicroBenchmark
    @OutputTimeUnit(TimeUnit.SECONDS)
    public List<String> parallel(BenchmarkState state) {
        return
            Arrays.stream(state.array)
                  .parallel()
                  .map(x -> x.toLowerCase())
                  .collect(Collectors.toList());
    }
}

我使用命令运行它:

java -jar dist/microbenchmarks.jar ".*SO23170832.*" -wi 5 -i 5 -f 1

(这些选项表示五次预热迭代,五次基准迭代和一次分叉JVM.)在运行期间,JMH发出了许多冗长的消息,我已经省略了.总结结果如下.

Benchmark                       Mode   Samples         Mean   Mean error    Units
c.s.q.SO23170832.parallel      thrpt         5        4.600        5.995    ops/s
c.s.q.SO23170832.sequential    thrpt         5        1.500        1.727    ops/s

请注意,结果是以每秒操作数为单位,因此看起来并行运行速度比顺序运行快三倍.但我的机器只有两个核心.嗯.每次运行的平均错误实际上大于平均运行时间!WAT？这里有些可疑的东西.

这给我们带来了第三个问题.仔细查看工作负载,我们可以看到它为每个输入分配一个新的String对象,并且还将结果收集到一个列表中,这涉及大量的重新分配和复制.我猜这会导致相当数量的垃圾收集.我们可以通过在启用GC消息的情况下重新运行基准来看到这一点:

java -verbose:gc -jar dist/microbenchmarks.jar ".*SO23170832.*" -wi 5 -i 5 -f 1

这给出了如下结果:

[GC (Allocation Failure)  512K->432K(130560K), 0.0024130 secs]
[GC (Allocation Failure)  944K->520K(131072K), 0.0015740 secs]
[GC (Allocation Failure)  1544K->777K(131072K), 0.0032490 secs]
[GC (Allocation Failure)  1801K->1027K(132096K), 0.0023940 secs]
# Run progress: 0.00% complete, ETA 00:00:20
# VM invoker: /Users/src/jdk/jdk8-b132.jdk/Contents/Home/jre/bin/java
# VM options: -verbose:gc
# Fork: 1 of 1
[GC (Allocation Failure)  512K->424K(130560K), 0.0015460 secs]
[GC (Allocation Failure)  933K->552K(131072K), 0.0014050 secs]
[GC (Allocation Failure)  1576K->850K(131072K), 0.0023050 secs]
[GC (Allocation Failure)  3075K->1561K(132096K), 0.0045140 secs]
[GC (Allocation Failure)  1874K->1059K(132096K), 0.0062330 secs]
# Warmup: 5 iterations, 1 s each
# Measurement: 5 iterations, 1 s each
# Threads: 1 thread, will synchronize iterations
# Benchmark mode: Throughput, ops/time
# Benchmark: com.stackoverflow.questions.SO23170832.parallel
# Warmup Iteration   1: [GC (Allocation Failure)  7014K->5445K(132096K), 0.0184680 secs]
[GC (Allocation Failure)  7493K->6346K(135168K), 0.0068380 secs]
[GC (Allocation Failure)  10442K->8663K(135168K), 0.0155600 secs]
[GC (Allocation Failure)  12759K->11051K(139776K), 0.0148190 secs]
[GC (Allocation Failure)  18219K->15067K(140800K), 0.0241780 secs]
[GC (Allocation Failure)  22167K->19214K(145920K), 0.0208510 secs]
[GC (Allocation Failure)  29454K->25065K(147456K), 0.0333080 secs]
[GC (Allocation Failure)  35305K->30729K(153600K), 0.0376610 secs]
[GC (Allocation Failure)  46089K->39406K(154624K), 0.0406060 secs]
[GC (Allocation Failure)  54766K->48299K(164352K), 0.0550140 secs]
[GC (Allocation Failure)  71851K->62725K(165376K), 0.0612780 secs]
[GC (Allocation Failure)  86277K->74864K(184320K), 0.0649210 secs]
[GC (Allocation Failure)  111216K->94203K(185856K), 0.0875710 secs]
[GC (Allocation Failure)  130555K->114932K(199680K), 0.1030540 secs]
[GC (Allocation Failure)  162548K->141952K(203264K), 0.1315720 secs]
[Full GC (Ergonomics)  141952K->59696K(159232K), 0.5150890 secs]
[GC (Allocation Failure)  105613K->85547K(184832K), 0.0738530 secs]
1.183 ops/s

注意:以...开头的行#是正常的JMH输出行.所有其余的都是GC消息.这只是五次预热迭代中的第一次,在五次基准迭代之前.在其余迭代期间,GC消息以相同的方式继续.我认为可以肯定地说,测量的性能主要受GC开销的影响,并且不应该相信所报告的结果.

此时还不清楚该怎么做.这纯粹是一种综合工作量.与分配和复制相比,它显然只需要很少的CPU时间来完成实际工作.很难说你在这里想要衡量的是什么.一种方法是提出一种在某种意义上更"真实"的不同工作量.另一种方法是在基准测试期间更改堆和GC参数以避免GC.

Answer 2

在进行基准测试时,您应该注意JIT编译,并且基于JIT编译代码路径的数量,时序行为可以改变.如果我在测试程序中添加预热阶段,则并行版本比顺序版本快一点.结果如下:

Warmup...
Benchmark...
Run 0:  sequential 0.12s  -  parallel 0.11s
Run 1:  sequential 0.13s  -  parallel 0.08s
Run 2:  sequential 0.15s  -  parallel 0.08s
Run 3:  sequential 0.12s  -  parallel 0.11s
Run 4:  sequential 0.13s  -  parallel 0.08s

以下代码片段包含我用于此测试的完整源代码.

public static void main(String... args) {
    String[] array = new String[1000000];
    Arrays.fill(array, "AbabagalamagA");
    System.out.println("Warmup...");
    for (int i = 0; i < 100; ++i) {
        sequential(array);
        parallel(array);
    }
    System.out.println("Benchmark...");
    for (int i = 0; i < 5; ++i) {
        System.out.printf("Run %d:  sequential %s  -  parallel %s\n",
            i,
            test(() -> sequential(array)),
            test(() -> parallel(array)));
    }
}
private static void sequential(String[] array) {
    Arrays.stream(array).map(String::toLowerCase).collect(Collectors.toList());
}
private static void parallel(String[] array) {
    Arrays.stream(array).parallel().map(String::toLowerCase).collect(Collectors.toList());
}
private static String test(Runnable runnable) {
    long start = System.currentTimeMillis();
    runnable.run();
    long elapsed = System.currentTimeMillis() - start;
    return String.format("%4.2fs", elapsed / 1000.0);
}

Answer 3

使用多个线程处理数据有一些初始设置成本,例如初始化线程池.这些成本可能超过使用这些线程的收益,特别是如果运行时已经非常低.另外,如果存在争用,例如其他线程运行,后台进程等,则并行处理的性能可以进一步降低.

这个问题并不是并行处理的新问题.本文根据Java 8提供了一些细节,parallel()还有一些需要考虑的事项:http://java.dzone.com/articles/think-twice-using-java-8