Stream.reduce()和Stream.collect()之间惊人的性能差异
我想比较两种Java8流终端操作reduce()和collect()的并行性能。
让我们看看下面的Java8并行流示例：
import java.math.BigInteger;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Stream;
import static java.math.BigInteger.ONE;
public class StartMe {
static Function<Long, BigInteger> fac;
static {
fac = x -> x==0? ONE : BigInteger.valueOf(x).multiply(fac.apply(x - 1));
}
static long N = 2000;
static Supplier<BigInteger[]> one() {
BigInteger[] result = new BigInteger[1];
result[0] = ONE;
return () -> result;
}
static BiConsumer<BigInteger[], ? super BigInteger> accumulator() {
return (BigInteger[] ba, BigInteger b) -> {
synchronized (fac) {
ba[0] = ba[0].multiply(b);
}
};
}
static BiConsumer<BigInteger[], BigInteger[]> combiner() {
return (BigInteger[] b1, BigInteger[] b2) -> {};
}
public static void main(String[] args) throws Exception {
long t0 = System.currentTimeMillis();
BigInteger result1 = Stream.iterate(ONE, x -> x.add(ONE)).parallel().limit(N).reduce(ONE, BigInteger::multiply);
long t1 = System.currentTimeMillis();
BigInteger[] result2 = Stream.iterate(ONE, x -> x.add(ONE)).parallel().limit(N).collect(one(), accumulator(), combiner());
long t2 = System.currentTimeMillis();
BigInteger result3 = fac.apply(N);
long t3 = System.currentTimeMillis();
System.out.println("reduce(): deltaT = " + (t1-t0) + "ms, result 1 = " + result1);
System.out.println("collect(): deltaT = " + (t2-t1) + "ms, result 2 = " + result2[0]);
System.out.println("recursive: deltaT = " + (t3-t2) + "ms, result 3 = " + result3);
}
}它计算n！使用一些--当然很奇怪;-) -算法。
然而，性能结果却令人惊讶：
reduce(): deltaT = 44ms, result 1 = 3316275...
collect(): deltaT = 22ms, result 2 = 3316275...
recursive: deltaT = 11ms, result 3 = 3316275...几点意见：
- 我必须同步
accumulator()，因为它并行地访问相同的数组。 - 我预计
reduce()和collect()会产生同样的性能，但是reduce()比collect()慢2倍，即使collect()必须同步！ - 最快的算法是顺序和递归算法(这可能显示并行流管理的巨大开销)。
我没想到reduce()的表现会比collect()差，为什么会这样呢？
回答 1
Stack Overflow用户
发布于 2015-03-11 17:31:42
基本上，您是在测量第一次执行的代码的初始开销。不仅优化器还没有工作，您还在测量加载、验证和初始化类的开销。
因此，难怪评估时间会减少，因为每个评估都可以重用已经为之前的评估加载的类。在一个循环中运行所有三个评估，甚至只是更改顺序，都会给出一个完全不同的画面。
唯一可预测的结果是，简单的递归计算将具有最小的初始开销，因为它不需要加载Stream API类。
如果您多次运行代码，或者更好地使用一个复杂的基准测试工具，我想您将获得类似于我的结果，其中reduce的性能明显优于collect，而且确实比单线程方法更快。
collect速度慢的原因是您使用它完全错误。将查询每个线程的Supplier以获得一个不同的容器，因此累加器函数不需要任何额外的同步。但是，对于将不同线程的结果容器连接到一个结果中，组合器函数的正确工作是很重要的。
正确的做法是：
BigInteger[] result2 = Stream.iterate(ONE, x -> x.add(ONE)).parallel().limit(N)
.collect(()->new BigInteger[]{ONE},
(a,v)->a[0]=a[0].multiply(v), (a,b)->a[0]=a[0].multiply(b[0]));在我的系统中，它的性能与reduce方法相当。由于使用数组作为可变容器不能改变BigInteger的不可变性质，在这里使用collect没有任何好处，使用reduce是直接的，而且，正如所述，当两种方法都正确使用时，具有同等的性能。
顺便说一句，我不明白为什么这么多程序员试图创建自引用lambda表达式。递归函数的直进方式仍然是一种方法：
static BigInteger fac(long x) {
return x==0? ONE : BigInteger.valueOf(x).multiply(fac(x - 1));
}
static final Function<Long, BigInteger> fac=StartMe::fac;(尽管在您的代码中，根本不需要Function<Long, BigInteger>，只需直接调用fac(long) )。
最后，Stream.iterate和Stream.limit对于并行执行来说都是非常糟糕的。使用具有可预测大小和独立操作的流将显著优于您的解决方案：
BigInteger result4 = LongStream.rangeClosed(1, N).parallel()
.mapToObj(BigInteger::valueOf).reduce(BigInteger::multiply).orElse(ONE);https://stackoverflow.com/questions/28988262
复制相似问题
腾讯云开发者
