Stream的并行计算

  1 package com.cy.java8;
  2 
  3 import java.util.function.Function;
  4 import java.util.stream.LongStream;
  5 import java.util.stream.Stream;
  6 
  7 public class ParallelProcessing {
  8 
  9     public static void main(String[] args) {
 10         //查看计算机核心线程数
 11         //System.out.println(Runtime.getRuntime().availableProcessors());
 12 
 13         long fastest1 = measureSumPerformance(ParallelProcessing::normalAdd, 10000000);
 14         System.out.println("normalAdd the best processing time : " + fastest1 + " ms");
 15 
 16         long fastest2 = measureSumPerformance(ParallelProcessing::iterateStream, 10000000);
 17         System.out.println("iterateStream the best processing time : " + fastest2 + " ms");
 18 
 19         long fastest3 = measureSumPerformance(ParallelProcessing::parallelStream, 10000000);
 20         System.out.println("parallelStream the best processing time : " + fastest3 + " ms");
 21 
 22         long fastest4 = measureSumPerformance(ParallelProcessing::parallelStream2, 10000000);
 23         System.out.println("parallelStream2 the best processing time : " + fastest4 + " ms");
 24 
 25         long fastest5 = measureSumPerformance(ParallelProcessing::parallelStream3, 10000000);
 26         System.out.println("parallelStream3 the best processing time : " + fastest5 + " ms");
 27 
 28     }
 29 
 30 
 31     /**
 32      * 将下面的求和方法分别计算10次，取10次中运行最短的时间
 33      */
 34     private static long measureSumPerformance(Function<Long, Long> adder, long limit){
 35         long fastest = Long.MAX_VALUE;
 36 
 37         for(int i=0; i<10; i++){
 38             long startTime = System.currentTimeMillis();
 39             long result = adder.apply(limit);
 40             long spendTime = System.currentTimeMillis() - startTime;
 41             System.out.println("the sum result is " + result);
 42             if(spendTime < fastest){
 43                 fastest = spendTime;
 44             }
 45         }
 46         return fastest;
 47     }
 48 
 49     /**
 50      * 计算一串long类型的总和，普通的stream
 51      * @param limit
 52      * @return
 53      */
 54     private static long iterateStream(long limit){
 55         return Stream.iterate(1L, i->i+1).limit(limit).reduce(0L, Long::sum);
 56     }
 57 
 58     /**
 59      * 使用Stream.parallel
 60      * 比较慢，为什么？
 61      * Stream.iterate不适合并行计算
 62      */
 63     private static long parallelStream(long limit){
 64         return Stream.iterate(1L, i->i+1).parallel().limit(limit).reduce(0L, Long::sum);
 65     }
 66 
 67     /**
 68      * 将上面的Stream先自动拆箱为long，再并行
 69      * 虽然拆箱为LongStream，还是很慢，为什么？
 70      * Stream.iterate不适合并行计算
 71      */
 72     private static long parallelStream2(long limit){
 73         return Stream.iterate(1L, i -> i + 1)
 74                         .mapToLong(Long::longValue)
 75                         .parallel().limit(limit).reduce(0L, Long::sum);
 76     }
 77 
 78     /**
 79      * 使用LongStream.range
 80      * 很快，比normalAdd快了近一倍，为什么？
 81      * LongStream、IntStream等..它们的IntStream.range非常卓越的适合并行计算
 82      */
 83     private static long parallelStream3(long limit){
 84         return LongStream.rangeClosed(1, limit).parallel().reduce(0L, Long::sum);
 85     }
 86 
 87 
 88     /**
 89      * 以前的写法
 90      * @param limit
 91      * @return
 92      */
 93     private static long normalAdd(long limit){
 94         long result = 0L;
 95         for(long i=0L; i <= limit; i++){
 96             result += i;
 97         }
 98         return result;
 99     }
100 }