Sha*_*riq 5 performance java-8 java-stream rx-java2
我已经成为Java 8中Java函数编程以及Rx java的忠实粉丝.但是一位同事最近指出,使用这些产品会影响性能.因此决定运行JMH Bench标记,但看起来他是对的.无论我做什么,我都无法获得流版本以提供更好的性能.以下是我的代码
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@BenchmarkMode(Mode.AverageTime)
@OperationsPerInvocation(StreamVsVanilla.N)
public class StreamVsVanilla {
public static final int N = 10000;
static List<Integer> sourceList = new ArrayList<>(N);
static {
for (int i = 0; i < N; i++) {
sourceList.add(i);
}
}
@Benchmark
public List<Double> vanilla() {
List<Double> result = new ArrayList<Double>(sourceList.size() / 2 + 1);
for (Integer i : sourceList) {
if (i % 2 == 0){
result.add(Math.sqrt(i));
}
}
return result;
}
@Benchmark
public List<Double> stream() {
return sourceList.stream().parallel()
.mapToInt(Integer::intValue)
.filter(i -> i % 2 == 0)
.mapToDouble(i->(double)i)
.map(Math::sqrt)
.boxed()
.collect(Collectors.toList());
}
@Benchmark
public List<Double> rxjava2(){
return Flowable.fromIterable(sourceList)
.parallel()
.runOn(Schedulers.computation())
.filter(i->i%2==0)
.map(Math::sqrt)
.collect(()->new ArrayList<Double>(sourceList.size()/2+1),ArrayList::add)
.sequential()
.blockingFirst();
}
public static void main(String[] args) throws RunnerException {
Options options = new OptionsBuilder()
.include(StreamVsVanilla.class.getSimpleName()).threads(1)
.forks(1).shouldFailOnError(true).shouldDoGC(true)
.jvmArgs("-server").build();
new Runner(options).run();
}
}
以上代码的结果:
# Run complete. Total time: 00:03:16
Benchmark Mode Cnt Score Error Units
StreamVsVanilla.rxjava2 avgt 20 1179.733 ± 322.421 ns/op
StreamVsVanilla.stream avgt 20 10.556 ± 1.195 ns/op
StreamVsVanilla.vanilla avgt 20 8.220 ± 0.705 ns/op
即使我删除了parellal运算符并使用如下的顺序版本:
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@BenchmarkMode(Mode.AverageTime)
@OperationsPerInvocation(StreamVsVanilla.N)
public class StreamVsVanilla {
public static final int N = 10000;
static List<Integer> sourceList = new ArrayList<>(N);
static {
for (int i = 0; i < N; i++) {
sourceList.add(i);
}
}
@Benchmark
public List<Double> vanilla() {
List<Double> result = new ArrayList<Double>(sourceList.size() / 2 + 1);
for (Integer i : sourceList) {
if (i % 2 == 0){
result.add(Math.sqrt(i));
}
}
return result;
}
@Benchmark
public List<Double> stream() {
return sourceList.stream()
.mapToInt(Integer::intValue)
.filter(i -> i % 2 == 0)
.mapToDouble(i->(double)i)
.map(Math::sqrt)
.boxed()
.collect(Collectors.toList());
}
@Benchmark
public List<Double> rxjava2(){
return Observable.fromIterable(sourceList)
.filter(i->i%2==0)
.map(Math::sqrt)
.collect(()->new ArrayList<Double>(sourceList.size()/2+1),ArrayList::add)
.blockingGet();
}
public static void main(String[] args) throws RunnerException {
Options options = new OptionsBuilder()
.include(StreamVsVanilla.class.getSimpleName()).threads(1)
.forks(1).shouldFailOnError(true).shouldDoGC(true)
.jvmArgs("-server").build();
new Runner(options).run();
}
}
结果不是很有利:
# Run complete. Total time: 00:03:16
Benchmark Mode Cnt Score Error Units
StreamVsVanilla.rxjava2 avgt 20 12.226 ± 0.603 ns/op
StreamVsVanilla.stream avgt 20 13.432 ± 0.858 ns/op
StreamVsVanilla.vanilla avgt 20 7.678 ± 0.350 ns/op
有人可以帮我弄清楚mi做错了什么吗？
编辑:
akarnokd指出我在顺序版本中使用额外的阶段到我的流版本中的unbox和box(我添加它以避免在过滤器和地图方法中隐式装箱拆箱)但是它变慢了所以我尝试没有那些代码如下
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@BenchmarkMode(Mode.AverageTime)
@OperationsPerInvocation(StreamVsVanilla.N)
public class StreamVsVanilla {
public static final int N = 10000;
static List<Integer> sourceList = new ArrayList<>(N);
static {
for (int i = 0; i < N; i++) {
sourceList.add(i);
}
}
@Benchmark
public List<Double> vanilla() {
List<Double> result = new ArrayList<Double>(sourceList.size() / 2 + 1);
for (Integer i : sourceList) {
if (i % 2 == 0){
result.add(Math.sqrt(i));
}
}
return result;
}
@Benchmark
public List<Double> stream() {
return sourceList.stream()
.filter(i -> i % 2 == 0)
.map(Math::sqrt)
.collect(Collectors.toList());
}
@Benchmark
public List<Double> rxjava2(){
return Observable.fromIterable(sourceList)
.filter(i->i%2==0)
.map(Math::sqrt)
.collect(()->new ArrayList<Double>(sourceList.size()/2+1),ArrayList::add)
.blockingGet();
}
public static void main(String[] args) throws RunnerException {
Options options = new OptionsBuilder()
.include(StreamVsVanilla.class.getSimpleName()).threads(1)
.forks(1).shouldFailOnError(true).shouldDoGC(true)
.jvmArgs("-server").build();
new Runner(options).run();
}
}
结果仍然或多或少相同:
# Run complete. Total time: 00:03:16
Benchmark Mode Cnt Score Error Units
StreamVsVanilla.rxjava2 avgt 20 10.864 ± 0.555 ns/op
StreamVsVanilla.stream avgt 20 10.466 ± 0.050 ns/op
StreamVsVanilla.vanilla avgt 20 7.513 ± 0.136 ns/op
启动多个线程并将值分派给多个线程的成本相对较高。为了弥补这一点，并行计算的成本通常比基础设施开销高几倍。然而，对于 RxJava 中的情况，Math::sqrt 是如此微不足道，并行开销决定了性能。
那么为什么 Stream 快了两个数量级呢？我只能假设线程窃取发生在基准线程完成大部分实际工作的情况下，也许一个后台线程完成了少量剩余工作，因为当后台线程旋转时，主线程已经窃取了大部分任务后退。因此，您没有像 RxJava 的并行那样严格的并行执行，其中操作员以循环方式分派工作，以便所有并行轨道可以大致同等地变得忙碌。
我认为 Stream 版本中有额外的拆箱和装箱阶段会增加一点开销。尝试不使用它：
return sourceList.stream()
.filter(i -> i % 2 == 0)
.map(Math::sqrt)
.collect(Collectors.toList());