为什么并行流更慢？

Question

我在测试无限流时，写了个基准测试程序。提供的数字越大，程序完成的速度就越快。但是，我惊讶地发现使用并行流比顺序流的性能要差得多。直觉上，人们会认为在多线程环境下生成和评估随机数的无限流会更快，但事实似乎并不是这样。为什么会这样呢？

    final int target = Integer.parseInt(args[0]);
    if (target <= 0) {
        System.err.println("Target must be between 1 and 2147483647");
        return;
    }

    final long startTime, endTime;
    startTime = System.currentTimeMillis();

    System.out.println(
        IntStream.generate(() -> new Double(Math.random()*2147483647).intValue())
        //.parallel()
        .filter(i -> i <= target)
        .findFirst()
        .getAsInt()
    );

    endTime = System.currentTimeMillis();
    System.out.println("Execution time: "+(endTime-startTime)+" ms");

- Sina Madani

---

将琐碎的任务并行化总是会更慢。多线程有足够的开销，任务需要证明其成本可行，否则你将看不到任何收益。此外，一次测试没有意义，至少要将其放入循环中并取平均值。 - Carcigenicate

@Carcigenicate 除此之外，我认为 Math.random() 会减慢它的速度 :-) 如果感兴趣，请查看我的答案。 - Lachezar Balev

你知道有一个内置的 Random.ints() 流可以获取一串随机数吗？ - Louis Wasserman

你的例子涉及了一些并行化失败的风险因素。首先，findFirst() 与流的遭遇顺序绑定，这会妨碍库进行并行化（你应该使用 findAny() 替代）。其次，每个元素所执行的工作量较少，这意味着创建、调度和协调任务的开销可能会压倒实际工作量。此外，每个元素的工作量也没有表现出很多的局部性。因此，并行加速的条件并不是很成熟。（另外，你的基准测试方法很可能会得到毫无意义的数字。） - Brian Goetz

@Holger 是的， generate() 函数的结果是无序的。但并行和遇到顺序之间的关系是微妙的，并且经常导致无法很好地并行化。因此，即使它不适用于这个特定代码片段，我觉得提到这种联系也是值得的。 - Brian Goetz

Answer 1

我完全同意其他评论和答案，但确实你的测试在目标非常低的情况下表现奇怪。在我的平凡笔记本上，当给出非常低的目标时，并行版本的平均速度大约慢了60倍。这种极端差异无法通过流API中并行化的开销来解释，所以我也感到惊讶。我认为罪魁祸首就在这里：

Math.random()

这个调用在内部依赖于一个全局实例java.util.Random。在Random文档中写到：

java.util.Random的实例是线程安全的。然而，在跨线程使用同一java.util.Random实例时，可能会遇到争用和随之而来的性能问题。在多线程设计中，考虑使用ThreadLocalRandom代替。

因此，我认为并行执行与顺序执行相比真正的低性能是由于随机数中的线程争用而不是其他任何开销所解释的。如果您改用文档中推荐的ThreadLocalRandom，那么性能差异就不会那么明显。另一个选项是实现更高级的数字供应商。

- Lachezar Balev

---

查看Math.random()的内部实现 - 它使用原子操作完成，没有任何锁。很难相信它会引入这样的差异。 - Sergey Fedorov

是的，我刚刚检查了一下。但它确实与线程本地随机数引入了巨大的差异，就像文档所述一样... :O - Lachezar Balev

@SergeyFedorov，虽然原子操作没有锁，但在原子操作的do-while循环中，直到状态被认为一致之前，它们可以极大地降低性能。问题在于Unsafe.compareAndSwap() - 它们在内部使用繁忙等待。在高争用情况下，成本呈指数级增长，最好使用一组操作的同步块。只有测试才能确定最佳策略。当然，ThreadLocalRandom在两种方法中每次都会获胜。 - Alex Pakka

你确定CAS操作会使用等待吗？有任何证据吗？ - Sergey Fedorov

在并行流操作的上下文中，你不应该使用 ThreadLocalRandom，而是应该使用 new SplittableRandom().ints(0, 2147483647).parallel() … 来创建流。 - Holger

@Sergey Fedorov：CAS不使用wait，但是如果CAS失败，则调用代码必须重试操作（除非失败是一种选项）。重复的重试类似于轮询，这就是Alex Pakka所说的“busy wait”。它不是将线程置于等待状态的意义上的等待。 - Holger

Answer 2

将工作传递给多个线程的成本很高，尤其是第一次这样做。这种成本相当固定，因此即使您的任务很简单，开销也相对较高。

你面临的问题之一是高度低效的代码是确定解决方案性能的非常糟糕的方式。此外，它第一次运行和几秒钟后运行的方式通常可以相差100倍（甚至更多）。我建议使用已经优化的示例，然后再尝试使用多个线程。

例如：

long start = System.nanoTime();
int value = (int) (Math.random() * (target+1L));
long time = System.nanoTime() - value;
// don't time IO as it is sooo much slower
System.out.println(value);

注意：在代码预热并编译之前，这不会很有效。即忽略此代码运行的前2-5秒钟。

- Peter Lawrey

Answer 3

根据各种答案的建议，我认为我已经解决了问题。虽然我不确定瓶颈是什么，但在i5-4590T上，使用以下代码的并行版本比顺序变体执行得更快。为简洁起见，我只包含了（重构后）代码的相关部分：

static IntStream getComputation() {
    return IntStream
            .generate(() -> ThreadLocalRandom.current().nextInt(2147483647));
}

static void computeSequential(int target) {
    for (int loop = 0; loop < target; loop++) {
        final int result = getComputation()
                    .filter(i -> i <= target)
                    .findAny()
                    .getAsInt();
        System.out.println(result);
    }
}

static void computeParallel(int target) {
     IntStream.range(0, target)
                .parallel()
                .forEach(loop -> {
                    final int result = getComputation()
                        .parallel()
                        .filter(i -> i <= target)
                        .findAny()
                        .getAsInt();
                    System.out.println(result);
                });
}

编辑：我还应该注意到，我把它们都放在一个循环中以获得更长的运行时间。

- Sina Madani

---

这不是一个有效的测试，因为您现在正在另一个并行操作内执行并行流操作，这将使内部操作（即您问题的实际操作）基本上成为顺序操作，因为工作以完全不同的方式分割，即范围IntStream.range(0, target)将被并行处理，而不是随机数序列。 - Holger

@Holger 但是在多台机器上测试后，我仍然得到了我期望的结果。在四核CPU上，对于大多数小于1000的数字，并行版本的性能比顺序版本快3倍以上。 - Sina Madani

事实上，我尝试将.parallel()添加到顺序版本中，但速度仍然较慢，所以结果表明加速来自InstStream.range而不是随机数流。 - Sina Madani