用parallelism=1实现串行和并行执行的区别

Question

请给我参考一下为什么使用Java的以下两个阶乘实现在执行时间上存在显著差异：

1. 串行实现
2. 并行实现(使用Stream.parallel())在自定义的叉连接池中执行，并行性设置为1

我的期望值是接近执行时间，但是并行版本的速度增加了2倍。我没有运行任何专门的基准测试，但是即使在冷启动jvm中，执行时间也不应该有太大的差异。Bellow I附加了这两种实现的源代码：

• 并行

public class FastFactorialSupplier implements FactorialSupplier {
  private final ExecutorService executorService;

  public FastFactorialSupplier(ExecutorService executorService) {
      this.executorService = executorService;
  }

  @Override
  public BigInteger get(long k) {
      try {
          return executorService
                  .submit(
                          () -> LongStream.range(2, k + 1)
                                  .parallel()
                                  .mapToObj(BigInteger::valueOf)
                                  .reduce(BigInteger.ONE, (current, factSoFar) -> factSoFar.multiply(current))
                  )
                  .get();
      } catch (InterruptedException | ExecutionException e) {
          e.printStackTrace();
      }

      return BigInteger.ZERO;
  }
}

• 串行

public class MathUtils {

  public static BigInteger factorial(long k) {
      return LongStream.range(2, k + 1)
              .mapToObj(BigInteger::valueOf)
              .reduce(BigInteger.ONE, (current, factSoFar) -> factSoFar.multiply(current));
  }
}

下面是带有附加示例执行时间的测试用例，作为基于intellij运行程序显示的注释。

    @Test
    public void testWithoutParallel() {
        //2s 403
        runTest(new DummyFactorialSupplier()); // uses MathUtils.factorial
    }

    @Test
    public void testParallelismWorkStealing1() {
        //1s 43
        runTest(new FastFactorialSupplier(Executors.newWorkStealingPool(1)));
    }

    @Test
    public void testParallelismForkJoin1() {
        // 711ms
        runTest(new FastFactorialSupplier(new ForkJoinPool(1)));
    }

    @Test
    public void testExecutorForkJoin() {
        //85ms
        runTest(new FastFactorialSupplier(new ForkJoinPool()));
    }

    private void runTest(FactorialSupplier factorialSupplier) {
        BigInteger result = factorialSupplier.get(100000);
        assertNotNull(result);
//        assertEquals(456574, result.toString().length());
    }

这些测试是使用java 11运行的，因为java 8中存在自定义叉连接池- https://bugs.openjdk.java.net/browse/JDK-8190974的问题。

是否可以进行与伪并行处理相关的优化，以及如何调度执行，而不存在这样的优化，因为执行完全是顺序的？

编辑：

我还使用jmh运行微基准测试。

平行：

public class FastFactorialSupplierP1Test {

    @Benchmark
    @BenchmarkMode({Mode.AverageTime, Mode.SampleTime, Mode.SingleShotTime, Mode.Throughput, Mode.All})
    @Fork(value = 1, warmups = 1)
    public void measure() {
        runTest(new FastFactorialSupplier(new ForkJoinPool(1)));
    }

    private void runTest(FactorialSupplier factorialSupplier) {
        BigInteger result = factorialSupplier.get(100000);
        assertNotNull(result);
    }

    public static void main(String[] args) throws Exception {
        org.openjdk.jmh.Main.main(args);
    }
}

序列号：

public class SerialFactorialSupplierTest {
    @Benchmark
    @BenchmarkMode({Mode.AverageTime, Mode.SampleTime, Mode.SingleShotTime, Mode.Throughput, Mode.All})
    @Fork(value = 1, warmups = 1)
    public void measure() {
        runTest(new DummyFactorialSupplier());
    }

    private void runTest(FactorialSupplier factorialSupplier) {
        BigInteger result = factorialSupplier.get(100000);
        assertNotNull(result);
    }

    public static void main(String[] args) throws Exception {
        org.openjdk.jmh.Main.main(args);
    }
}

public class IterativeFactorialTest {
    @Benchmark
    @BenchmarkMode({Mode.AverageTime, Mode.SampleTime, Mode.SingleShotTime, Mode.Throughput, Mode.All})
    @Fork(value = 1, warmups = 1)
    public void measure() {
        runTest(new IterativeFact());
    }

    private void runTest(FactorialSupplier factorialSupplier) {
        BigInteger result = factorialSupplier.get(100000);
        assertNotNull(result);
    }

    public static void main(String[] args) throws Exception {
        org.openjdk.jmh.Main.main(args);
    }

    class IterativeFact implements FactorialSupplier {

        @Override
        public BigInteger get(long k) {
            BigInteger result = BigInteger.ONE;

            while (k-- != 0) {
                result = result.multiply(BigInteger.valueOf(k));
            }

            return result;
        }
    }
}

结果：

FastFactorialSupplierP1Test.measure                    avgt    5  0.437 ± 0.006   s/op
IterativeFactorialTest.measure                         avgt    5  2.643 ± 0.383   s/op
SerialFactorialSupplierTest.measure                    avgt    5  2.226 ± 0.044   s/op

Answer 1

您选择了一个操作，其性能取决于评估的顺序。考虑到BigInteger.multiply的性能取决于这两个因素的大小。然后，使用累加值作为下一个乘法的一个因素，通过一个BigInteger实例序列运行将使操作越来越昂贵，得到的越远。

相反，当您将值的范围划分为较小的范围时，对每个范围分别执行乘法，并将范围的结果相乘，您将获得性能优势，即使这些子范围不是同时计算的。

因此，当并行流将工作分割成块时，可能会被其他工作线程捕获，但最终会在同一个线程中对它们进行评估，在此特定设置中，由于计算顺序的改变，您仍然可以获得性能改进。

我们可以通过删除所有与流和线程池相关的工件来测试这一点：

public static BigInteger multiplyAll(long from, long to, int split) {
    if(split < 1 || to - from < 2) return serial(from, to);
    split--;
    long middle = (from + to) >>> 1;
    return multiplyAll(from, middle, split).multiply(multiplyAll(middle, to, split));
}

private static BigInteger serial(long l1, long l2) {
    BigInteger bi = BigInteger.valueOf(l1++);
    for(; l1 < l2; l1++) {
        bi = bi.multiply(BigInteger.valueOf(l1));
    }
    return bi;
}

我手头没有JMH设置，可以发布可承受压力的结果，但简单的运行显示，大小顺序与您的结果相匹配，只是一个单独的拆分已经大致将评估时间减半，更高的数字仍然可以提高性能，尽管曲线变得更平坦。

正如在ForkJoinTask.html#getSurplusQueuedTaskCount()中解释的那样，将工作分开是一种合理的策略，因为每个工作人员都有几个额外的任务，可能被其他线程捕获，这可能补偿不平衡的工作负载，例如，如果某些元素比其他元素更便宜的话。显然，并行流没有特殊的代码来处理没有额外工作线程的情况，因此，您可以看到分割工作的效果，即使只有一个线程来处理它。