能否请您引用一下为什么使用 Java Stream API 的以下 2 个阶乘实现在执行时间上存在显着差异:

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

我的期望是接近执行时间，但是并行版本的速度提高了 2 倍。 我没有运行任何专门的基准测试，但是即使在冷启动 jvm 中，执行时间也不应该相差太多。 下面我附上两个实现的源代码:

• Parallel

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;
  }
}

• Serial

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 junit runner 显示内容的评论。

    @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 中存在自定义 fork 连接池问题 - 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

最佳答案

您选择了一个性能取决于评估顺序的操作。只需考虑 BigInteger.multiply 的性能取决于这两个因素的大小。然后，遍历一系列 BigInteger 实例，将累积值作为下一个乘法的因子将使操作变得越来越昂贵，你得到的越远。

相比之下，当您将值的范围拆分为更小的范围时，对每个范围单独执行乘法并将范围的结果相乘，您将获得性能优势，即使这些子范围没有同时计算。

因此，当并行流将工作拆分成多个 block ，可能被其他工作线程接收，但最终在同一个线程中评估它们时，您仍然可以获得性能改进，在这个特定的设置中，由于更改了评估顺序。

我们可以通过删除所有 Stream 和线程池相关的工件来测试它:

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() 中所述, 拆分工作是一个合理的策略，这样每个工作人员都有一些额外的任务，可能会被其他线程接管，这可能会补偿不平衡的工作负载，例如如果某些元素的处理成本低于其他元素。显然，并行流没有特殊的代码来处理没有额外的工作线程的情况，因此，即使只有一个线程来处理它，您也可以看到拆分工作的效果。

关于java - 并行度为 1 的串行和并行执行之间的区别，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/56418666/