153、Java8新特性之并行流和串行流

并行流就是把一个内容分成多个数据块，并用不同的 线程 分别处理每个数据块的流。Java8中将并行做了优化，我们可以很容易的对数据进行并行操作。StreamAPI可以声明性的通过parallel()与sequential()在并行流与顺序流之间进行切换。

再说并行流之前，我们先了解一个框架Fork/Join：Fork/Join框架就是在必要情况下，将一个大的任务进行拆分fork成若干个小的任务，知道不可再拆分，然后再将一个个小的任务的运算结果进行join汇总。

​

Fork/Join框架使用起来比较复杂，但是效率高。Fork/Join框架采取“工作窃取”模式（work-stealing）：当执行新的任务时，它可以将其拆分为更小的任务执行，并将小任务加入到线程队列中，然后再从一个随机线程的队列中偷一个并把它放在自己的队列中。

相对于一般的 线程池 实现，Fork/Join框架的优势体现在对其中包含的任务的处理方式上。在一般的线程池中，如果一个线程正在执行的任务由于某些原因无法继续执行，那么该线程会处于等待状态，而在Fork/Join框架实现中，如果某个子问题由于等待另外一个子问题的完成而无法继续运行，那么处理该子问题的线程会主动寻找其他尚未运行的子问题来执行。这种方式减少了线程的等待时间，提高了性能。

Fork/Join模式使用：

* 1.继承 抽象类 RecusiveTask（有返回值，泛型指定返回值）或者抽象类RecusiveAction（无返回值）

* 2.实现其中的compute方法，以递归形式拆分任务

* 3.任务拆分完成之后，调用fork，压入线程队列

* 4.任务处理完成之后，调用join，返回值

* 5.运行程序时，需要ForkJoinPool的支持

* 5.1.新建ForkJoinPool

* 5.2.用父类ForkJoinTask去引用需要新建的对象

* 5.3.调用ForkJoinPool的invoke，进行fork/join，接收返回值

测试Fork/Join框架：

public class ForkJoinCalculate extends RecursiveTask {
 private long  start ;
 private  Long  end;
 private static final long THRESHOLD = 10000;
 public ForkJoinCalculate(long start, long end) {
 this.start = start;
 this.end = end;
 }
 @Override
 protected Long compute() {
 long length = end - start;
 if (length <= THRESHOLD){
 long sum = 0;
 for ( long  i = start; i <= end; i++){
 sum += i;
 }
 return sum;
 }else {
 long middle = (start + end) / 2;
 ForkJoinCalculate left = new ForkJoinCalculate(start,middle);
 left.fork();//拆分子任务，同时压入线程队列
 ForkJoinCalculate  right  = new ForkJoinCalculate(middle,end);
 right.fork();
 return (long)right.join() + (long)left.join();
 }
 }
}
 

public class TestForkJoin {
 //使用ForkJoin框架计算从0加到1亿的和
 @Test
 public void test1(){
 Instant start = Instant.now();
 ForkJoinPool pool = new ForkJoinPool();
 ForkJoinTask<Long> task = new ForkJoinCalculate(0,100000000L);
 Long sum = pool.invoke(task);
 System.out.println("和是：" + sum);
 Instant end = Instant.now();
 System.out.println("耗费时间是：" + Duration.between(start,end).toMillis() + "毫秒");
 }
 //使用普通for循环计算求和
 @Test
 public void test2(){
 Instant start = Instant.now();
 
 long sum = 0L;
 for (long i = 0; i <= 100000000L; i++){
 sum += i;
 }
 System.out.println("和是：" + sum);
 
 Instant end = Instant.now();
 System.out.println("耗费时间是：" + Duration.between(start,end).toMillis() + "毫秒");
 }
}
 

数据量越大，Fork/Join框架的优势越明显。

下面看Java8使用并行流怎么实现：

 //使用Java8并行流计算求和
 @Test
 public void test3(){
 Instant start = Instant.now();
 long sum = LongStream.rangeClosed(0, 100000000L).parallel().reduce(0, Long::sum);
 System.out.println("和是：" + sum);
 Instant end = Instant.now();
 System.out.println("耗费时间是：" + Duration.between(start,end).toMillis() + "毫秒");
 }