stream聚合parallelStream原理

 

 

背景：ForkJoinPool的优势在于，可以充分利用多cpu，多核cpu的优势，把一个任务拆分成多个“小任务”，把多个“小任务”放到多个处理器核心上并行执行；当多个“小任务”执行完成之后，再将这些执行结果合并起来即可。这种  “分治”  思想值得学习。

 

 

parallelStream  就是把后续的流操作一个对象一个线程去处理，后续的操作还是一样，线程的来源是forkjoinpool 其线程的数量来源于机器cpu的大小---工作在本地jvm执行

 

任务分小，但是执行任务的线程默认是cpu个数（1，拆分任务，2，固定线程池中执行）

 

threadpoolexcutor不能实现线程执行的等待合并，不能选择优先执行子任务，除非用futureTask，forkjoinpool用他自己的task就可以实现线程的结果合并，并且线程会主动切换

到未阻塞的任务(至于子任务线程的拆分由自定义的task通过递归实现)

 

forkjoinpool适用于拆分任务，更适用于有父子依赖关系的任务

 

分析与使用：

Java7 提供了ForkJoinPool来支持将一个任务拆分成多个“小任务”并行计算，再把多个“小任务”的结果合并成总的计算结果。

ForkJoinPool是ExecutorService的实现类，因此是一种特殊的线程池。

使用方法：创建了ForkJoinPool实例之后，就可以调用ForkJoinPool的submit(ForkJoinTask<T> task) 或invoke(ForkJoinTask<T> task)方法来执行指定任务了。

其中ForkJoinTask代表一个可以并行、合并的任务。ForkJoinTask是一个抽象类，它还有两个抽象子类：RecusiveAction和RecusiveTask。其中RecusiveTask代表有返回值的任务，而RecusiveAction代表没有返回值的任务。

 

 

 

 

 

 

 

下面的UML类图显示了ForkJoinPool、ForkJoinTask之间的关系：

 

 

 

 

    

 

 

 

实战：

 

 

   例1：

                   有 返 回 值

 

import java.time.Duration;

import java.time.Instant;

import java.util.concurrent.ForkJoinPool;

import java.util.concurrent.ForkJoinTask;

import java.util.concurrent.RecursiveTask;

import java.util.stream.LongStream;

 

import org.junit.Test;

 

public class TestForkJoinPool {

 

public static void main(String[] args) {

Instant start = Instant.now();

 

ForkJoinPool pool = new ForkJoinPool();

 

ForkJoinTask<Long> task = new ForkJoinSumCalculate(0L, 50000000000L);

 

Long sum = pool.invoke(task);

 

System.out.println(sum);

 

Instant end = Instant.now();

 

System.out.println("耗费时间为：" + Duration.between(start, end).toMillis());//166-1996-10590

}

 

@Test

public void test1(){

Instant start = Instant.now();

 

long sum = 0L;

 

for (long i = 0L; i <= 50000000000L; i++) {

sum += i;

}

 

System.out.println(sum);

 

Instant end = Instant.now();

 

System.out.println("耗费时间为：" + Duration.between(start, end).toMillis());//35-3142-15704

}

 

 

}

 

class ForkJoinSumCalculate extends RecursiveTask<Long>{

 

/**

* 

*/

private static final long serialVersionUID = -259195479995561737L;

 

private long start;

private long end;

 

private static final long THURSHOLD = 10000L;  //临界值

 

public ForkJoinSumCalculate(long start, long end) {

this.start = start;

this.end = end;

}

 

@Override

protected Long compute() {

long length = end - start;

 

if(length <= THURSHOLD){

long sum = 0L;

 

for (long i = start; i <= end; i++) {

sum += i;

}

 

return sum;

}else{

 

long middle = (start + end) / 2;

 

ForkJoinSumCalculate left = new ForkJoinSumCalculate(start, middle); //通过递归实现

left.fork(); //进行拆分，同时压入线程队列

 

ForkJoinSumCalculate right = new ForkJoinSumCalculate(middle+1, end);

right.fork(); //

 

return left.join() + right.join();

}

}

 

}

   根据结果可得，在数量超过一定的时候效率比普通的for循环明显，但是数较少的时候，优势体现不明显。

 

 

 

例2：

     无 返 回 值

 

import java.util.concurrent.ForkJoinPool;

import java.util.concurrent.RecursiveAction;

import java.util.concurrent.TimeUnit;

 

public class ForkJoinPoolAction {

    

    public static void main(String[] args) throws Exception{

        PrintTask task = new PrintTask(0, 300);

        //创建实例，并执行分割任务

        ForkJoinPool pool = new ForkJoinPool();

        pool.submit(task);

         //线程阻塞，等待所有任务完成

        pool.awaitTermination(2, TimeUnit.SECONDS);

        pool.shutdown();

    }

}

 

 

class PrintTask extends RecursiveAction{

    private static final int THRESHOLD = 50; //最多只能打印50个数

    private int start;

    private int end;

    

    

 

    public PrintTask(int start, int end) {

        super();

        this.start = start;

        this.end = end;

    }

 

 

 

    @Override

    protected void compute() {

        

        if(end - start < THRESHOLD){

            for(int i=start;i<end;i++){

                System.out.println(Thread.currentThread().getName()+"的i值："+i);

            }

        }else {

            int middle =(start+end)/2;

            PrintTask left = new PrintTask(start, middle);

            PrintTask right = new PrintTask(middle, end);

            //并行执行两个“小任务”

            left.fork();

            right.fork();

        }

        

    }

    

}

 

 

总结：

 

 

在Java 7中引入了一种新的线程池：ForkJoinPool。

     它同ThreadPoolExecutor一样，也实现了Executor和ExecutorService接口。它使用了一个无限队列来保存需要执行的任务，而线程的数量则是通过构造函数传入，如果没有向构造函数中传入希望的线程数量，那么当前计算机可用的CPU数量会被设置为线程数量作为默认值。

 

 ForkJoinPool主要用来使用分治法(Divide-and-Conquer Algorithm)来解决问题。典型的应用比如快速排序算法。

这里的要点在于，ForkJoinPool需要使用相对少的线程来处理大量的任务。

       比如要对1000万个数据进行排序，那么会将这个任务分割成两个500万的排序任务和一个针对这两组500万数据的合并任务。以此类推，对于500万的数据也会做出同样的分割处理，到最后会设置一个阈值来规定当数据规模到多少时，停止这样的分割处理。比如，当元素的数量小于10时，会停止分割，转而使用插入排序对它们进行排序。

 

       那么到最后，所有的任务加起来会有大概2000000+个。问题的关键在于，对于一个任务而言，只有当它所有的子任务完成之后，它才能够被执行。

 

      所以当使用ThreadPoolExecutor时，使用分治法会存在问题，因为ThreadPoolExecutor中的线程无法像任务队列中再添加一个任务并且在等待该任务完成之后再继续执行。而使用ForkJoinPool时，就能够让其中的线程创建新的任务，并挂起当前的任务，此时线程就能够从队列中选择子任务执行。

 

     以上程序的关键是fork()和join()方法。在ForkJoinPool使用的线程中，会使用一个内部队列来对需要执行的任务以及子任务进行操作来保证它们的执行顺序。

 

 

 

那么使用ThreadPoolExecutor或者ForkJoinPool，会有什么性能的差异呢？

 

    首先，使用ForkJoinPool能够使用数量有限的线程来完成非常多的具有父子关系的任务，比如使用4个线程来完成超过200万个任务。但是，使用ThreadPoolExecutor时，是不可能完成的，因为ThreadPoolExecutor中的Thread无法选择优先执行子任务，需要完成200万个具有父子关系的任务时，也需要200万个线程，显然这是不可行的。

 

ps：ForkJoinPool在执行过程中，会创建大量的子任务，导致GC进行垃圾回收，这些是需要注意的。

 

 

 

 

原文：https://blog.csdn.net/jsutdoit/article/details/81114144 

 https://www.jianshu.com/p/2472f53d6e2f