java - 进一步并行计算两个Java parallelStream()

Question

我有一个方法可以并行计算列表中数字的平方并将它们相加:

public static double sumSquared(List<Double> values) {
    return values
            .parallelStream()
            .mapToDouble(d -> d * d)
            .sum();
}

假设有几个 double 列表，如何进一步并行化，例如

double result = sumSquared(list0) + (sumSquared(list1) * sumSquared(list2));

有什么优雅的方法吗？我想为每个列表操作启动单独的线程并没有多大帮助，因为它们使用相同的线程池。它是否有意义，或者结果是否应该按上述方式计算？

Answer 1

问题确实是，即使 sumSquared() 并行处理数据，每个 sumSquared() 调用都是按顺序执行的。因此，列表仍然是一个接一个地处理的，如果其中一些列表太小而无法拆分到多个 CPU，则会浪费 CPU 处理时间。

为避免这种情况，请使用新的 CompletionStage 并行运行整个计算。/CompletableFuture在 Java 8 中引入:

// each sumSquared() is submitted immediately to the common pool
CompletionStage<Double> sumList0 = CompletableFuture.supplyAsync(() -> sumSquared(list0));
CompletionStage<Double> sumList1 = CompletableFuture.supplyAsync(() -> sumSquared(list1));
CompletionStage<Double> sumList2 = CompletableFuture.supplyAsync(() -> sumSquared(list2));

// as soon as sumList1 and sumList2 have both complete, their product is computed
CompletionStage<Double> prodSumList1and2 = sumList1.thenCombine(sumList2, (a, b) -> a * b);
// as soon as their product is computed and sumList0 is finished, the final sum is computed
CompletionStage<Double> result = sumList0.thenCombine(prodSumList1and2, Double::sum);

System.out.println(result.toCompletableFuture().get());

请注意，这使用公共(public)池，其大小默认设置为使用所有 CPU。因此，如果您只执行 CPU 密集型操作，使用自定义池可能不会提高性能，除非您有其他并行运行的作业。

例如，如果您的处理更多地受 I/O 限制或内存限制，情况可能会有所不同。