《Java8 实战》笔记——2.并行流ParallelStream&ForkJoin原理

青旅半醒 2023-09-30 10:33 57阅读 0赞

**第 7 章**

并行数据处理与性能——流内部的工作原理；

1.  性能比较测试

// 输出CPU的核心数
    
    @Test
    public void func_01() {
      System.out.println(Runtime.getRuntime().availableProcessors());//打印CPU核心数量-6核
    }

并行流和串行流的互相转换：

@Test
    public void func_02() {//串行/并行流的互相转换-例如求和，比较一下性能
      //1.转化的方法.parallel()和.sequential()
      Stream.of(1, 2, 3).parallel().filter(n -> n < 3).sequential().forEach(System.out::println);
    }

**\*性能小测试**

（1）生成随机数组-排序

@Test
    public void func_04() {
      //排序
      int[] arr = getArray(50000000);
      TraditionSort(arr);
      StreamSort(arr);
      ParellelStreamSort(arr);
    }
    
    static int[] getArray(int n){//随机生成一个长度为n的乱序序列
      int[] res = new int[n];
      Random r = new Random();
      for (int i = 0; i < n; i++) {
        res[i] = (r.nextInt(10000000));
      }
      return res;
    }
    
    static void TraditionSort(int[] arr) {//n个乱序自然数-排序
      long start = System.nanoTime();
      Arrays.sort(arr);
      long end = System.nanoTime();
      System.out.println("传统方法排序时间：" + (end - start) / (1000000) + "ms");
    }
    
    static void StreamSort(int[] arr) {//Stream
      long start = System.nanoTime();
      Stream.of(arr).sorted().count();//因为流是"惰性计算"的，为了终结流得到结果
      long end = System.nanoTime();
      System.out.println("Stream排序时间：" + (end - start) / (1000000) + "ms");
    }
    
    static void ParellelStreamSort(int[] arr) {//parallelStream
      long start = System.nanoTime();
      Stream.of(arr).parallel().sorted().count();
      long end = System.nanoTime();
      System.out.println("parallelStream排序时间：" + (end - start) / (1000000) + "ms");
    }

> 输出结果：  
> 传统方法排序时间：4134ms  
> Stream排序时间：32ms  
> ParellelStream排序时间：4ms

分析：在排序上，并行流>串行>>JDK优化后的单线程传统排序；(排序每一步的计算结果不依赖之前一步的结果，只需要把每个结果"合并"，有点像归并排序的思想)

（2）自然数序列前n项累加求和

@Test
    public void func_05() {
    //生成自然数前n项-求和
      TraditionIterate(100000000);
      StreamIterate(100000000);
      StreamIterate_2(100000000);
      ParellelStreamIterate(100000000);
    }
    
    static void TraditionIterate(long n) {//传统求前n个自然数的和
      long start = System.nanoTime();
      long res = 0;
      for (int i = 1; i < n; i++) {
        res += i;
      }
      long end = System.nanoTime();
      System.out.println("求前n个自然数序列之和" + "\n传统方法:" + res + " 计算时间：" + (end - start) / (1000000) + "ms");
    }
    
    static void StreamIterate(long n) {//Stream求前n个自然数的和
      long start = System.nanoTime();
      long res = Stream.iterate(0L, i -> i + 1).limit(n).reduce(0L, (i, j) -> (i + j));
      long end = System.nanoTime();
      System.out.println("Stream:" + res + " 计算时间：" + (end - start) / (1000000) + "ms");
    }
    
    static void ParellelStreamIterate(long n) {//ParellelStream求前n个自然数的和
      long start = System.nanoTime();
      long res = Stream.iterate(0L, i -> i + 1).limit(n).parallel().reduce(0L, Long::sum);
      long end = System.nanoTime();
      System.out.println("Parallel-Stream:" + res + " 计算时间：" + (end - start) / (1000000) + "ms");
    }
    
    static void StreamIterate_2(long n) {//LongStream求前n个自然数的和
      long start = System.nanoTime();
      // long res = LongStream.iterate(0L, i -> i + 1).limit(n).reduce(0L, Long::sum);
      long res = LongStream.range(1,n).reduce(0L, Long::sum);
      long end = System.nanoTime();
      System.out.println("Stream_02:" + res + " 计算时间：" + (end - start) / (1000000) + "ms");
    }

> 输出：  
> 求前n个自然数序列之和  
> 传统方法:4999999950000000 计算时间：37ms  
> Stream:4999999950000000 计算时间：1135ms  
> Stream\_02:4999999950000000 计算时间：42ms  
> Parallel-Stream:4999999950000000 计算时间：23587ms

**\*解释**：for循环使用基本类型，没有装箱，而LongStream避免了装箱，因此相对快，也看出了装箱对性能的影响也是很可观的！

第二点，为什么"并行计算"不仅没有性能提升，还要慢那么多！？因为计算难以划分模块，求和需要依赖之前的计算结果！所以，parallelStream不能滥用，要小心！

**针对上面的求和——引出：流的线程安全问题**

eg：  
采用一种新思路，不依赖每次的求和结果，定义一个累加器；

@Test
    public void func_06() {//求和的另一种思路:使用求和器对象来保存上一次的计算结果（状态）;
      long start = System.nanoTime();
      long res = 0;
      Adder adder = new Adder();
      LongStream.range(1,100000000).forEach(adder::add);//示例使用求和器对象来保存上一次的计算结果（状态），这样每次计算就不依赖前一次而是仅依赖这个累加器；
      res = adder.total;
      long end = System.nanoTime();
      System.out.println("求前n个自然数序列之和" + "\n使用求和器的方法:" + res + " 计算时间：" + (end - start) / (1000000) + "ms");
      //现在，使用并行流进行上面的计算,看看时间和结果，看看会发什什么；
      long start2 = System.nanoTime();
      long res2 = 0;
      Adder adder2 = new Adder();
      LongStream.range(1,100000000).parallel().forEach(adder2::add);//使用并行流；
      res2 = adder2.total;
      long end2 = System.nanoTime();
      System.out.println("求前n个自然数序列之和" + "\n使用求和器的方法:" + res2 + " 计算时间：" + (end2 - start2) / (1000000) + "ms");
    }
    static class Adder{//累加器——记录当前的结果；
      public long total = 0;//当前结果
      public void add(long value){total+=value;}
    }

> 输出结果：  
> 求前n个自然数序列之和  
> 使用求和器的方法:4999999950000000 计算时间：86ms

求前n个自然数序列之和  
使用求和器的方法:512183811469674 计算时间：34ms

解释：total+=value;不是原子操作，并行时候会发生线程安全问题；再使用并行流的时候不要尝试使用"共享的可变状态";

**3. 对使用并行流来提高性能的建议**

*  并行流不总是比串行流块（参考上面的第一种求和），快实在某种条件下的；
 *  注意自动装箱带来的性能损失；（数值类有专门的流 LongStream、IntStream、LongStream）；
 *  依赖元素顺序的操作，如limit、findFirst，使用并行流操作的代价很大；
 *  考虑流水线的总成本，N为元素综述，Q为一个元素通过流水线的处理成本，总成本估计为N\*Q；当Q很大时考虑使用并行流（N大Q小则不需要）；
 *  数据量小不推荐使用并行流，并行是需要开销的，并行流带来的性能提升是"扣除了自身的并行化开销"之后的；
 *  考虑流的源-对应的数据结构是否易于分解，例如ArrayList/数组，而LinkList则必须遍历；
 *  注意路中间的操作组合，不同的组合能达到相同的结果，但是性能可能不同；
 *  考虑终端操作中合并步骤的代价大小，太大则组合子流的代价可能会超过并行带来的性能提升；

### 分支/合并框架-ForkJoinPool (并行流的实现原理-可区别于CompletableFuture) ###

这个框架是用来实现并行流的，ForkJoinPool的优势在于，可以充分利用多cpu、多核cpu的优势，把一个任务拆分成多个“小任务”，把多个“小任务”放到多个处理器核心上并行执行；当多个“小任务”执行完成之后，再将这些执行结果合并起来即可。它是ExcutorService接口的实现，把子任务分给ForkJoinPool中的工作线程。

核心方法：compute()

// 伪代码：
    if（任务足够小/不可分）{
      顺序计算该任务
    }
    else {
      将任务拆成子任务；
      递归调用本方法：拆分每个子任务，等所有子任务执行完
      合并每个子任务的计算结果；
    }

**是不是很像归并排序？来复习一波代码！**

归并排序就是利用归并的思想实现排序。基本思想是假设初始序列有n个记录，可以将其看成n个子序列，然后两两归并，得到n/2个有序子序列，然后继续两两归并，直到最后得到一个长度为n的有序序列。这种排序方法称为2路归并排序。  
eg：

// 归并排序，将数组递归划分，最小为块1个元素，然后将2个块“归并”，再将2个有2个元素的“已经有序”块再“归并”，以此类推；
        private static int[] MergeSort(int[] arrTest01) {
            if (arrTest01 == null) {
                return null;
            }
            int result[] = arrTest01.clone();
            int len = arrTest01.length;
            //
            MergingSort(result, 0, len - 1);
            return result;
        }
    
        private static void MergingSort(int[] src, int start, int end) {
            if (start >= end) {//注意递归返回的条件
                return;
            }
            //划分；
            int mid = (start + end) >>> 1; //分别对前后快递归调用“归并排序”
            MergingSort(src, start, mid);
            MergingSort(src, mid + 1, end);
            //合并前后块；
            Merge(src, start, end, mid);
        }
    
        private static void Merge(int[] src, int start, int end, int mid) {
            int indexA = start;
            int indexB = mid + 1;
            int index = 0;
            int[] tempArr = new int[end - start + 1];
            while (indexA <= mid && indexB <= end) {
                if (src[indexA] < src[indexB]) {
                    tempArr[index++] = src[indexA++];
                } else {
                    tempArr[index++] = src[indexB++];
                }
            }
            //只会执行下面while中的一个
            while (indexA <= mid) {
                tempArr[index++] = src[indexA++];
            }
            while (indexB <= end) {
                tempArr[index++] = src[indexB++];
            }
            //将tempArr合并到原来result[]里面
            index = start;
            while (index <= end) {
                src[index] = tempArr[index - start];
                index++;
            }
        }

compute()方法的一个关键步骤，判断任务可划分——Spliterator接口，核心方法trySplit()；（实践中，这个接口不需要我们自己用，但是可以了解一下原理）

当前的Spliterator调用trySplit()会把一些元素（任务）划出去分给一个新产生的Spliterator，让他们并行执行；

### ForkJoinPool的一个有趣的概念**"工作窃取"** ###

试想一种情况，尽管我们可以把任务拆分，甚至在数量少拆分的尽可能均匀，但每个任务的执行时间是不可控的，意味着两个线程即使有相同任务数，他们的结束时间不一定相同；

怎么办？——"窃取任务"，即让每个线程为他的任务列表保存一个"双向链式队列"，当前任务每执行一个任务，就从"头"往下去找下一个任务执行，而某个早早完成了所有任务的线程就从其他有任务的线程的"任务队列尾巴"去"偷"走一个任务执行，这样直至所有任务都被执行完；

这样做的目的——在线程之间重新分配和平衡任务负载；

### \***这里再来重新分析一下ForkJoinPool的原理** ###

在多线程并发编程中，有时候会遇到将大任务分解成小任务再并发执行的场景。Java 8新增的ForkJoinPool很好的支持了这个问题。

ForkJoinPool是一种支持任务分解的线程池，当提交给他的任务“过大”，他就会按照预先定义的规则将大任务分解成小任务，多线程并发执行。 一般要配合可分解任务接口ForkJoinTask来使用，ForkJoinTask有两个实现它的抽象类：RecursiveAction（有返回值）和RecursiveTask（无返回值）。

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

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

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

**应用场景**：简单来说，如果你的问题能很容易分解成子问题，则比较适合ForkJoinPool。如适合CPU密集型的场景。

eg：一个简单的实例：计算1+2+…+100;并行计算

public class RecursiveTaskDemo extends RecursiveTask<Integer> {
    
        private int first;
        private int last;
    
        public RecursiveTaskDemo(int first, int last) {//RecursiveTask即为被ForkJoinPool执行的Task，它的compute方法会把计算划分成子任务（递归调用）
            this.first = first;
            this.last = last;
        }
    
        @Override
        protected Integer compute() {
            System.out.println(Thread.currentThread().getName() + " ... ");
            /**
            * 这里面要写自己的划分逻辑 构造ForkJoin
            */
            int sum = 0;
            // 拆分任务
            if (last - first <= 2) {
                // 计算
                for (int i = first; i <= last; i++) {
                    sum += i;
                }
            } else {
                /**
                * 类似于分支递归思想，拆分成两个子任务；
                */
                RecursiveTaskDemo demo01 = new RecursiveTaskDemo(first, (last + first) / 2);
                RecursiveTaskDemo demo02 = new RecursiveTaskDemo((last + first) / 2 + 1, last);
    
                // 执行
                demo01.fork();//fork是分叉的意思；执行fork方法，会将当前任务放入workQueue里面；立即返回自身；
                demo02.fork();
    
                Integer a = demo01.join();//join得到上一步任务在线程池中运行后的执行结果；
                Integer b = demo02.join();
    
                sum = a + b;//合并两个分支的结果
            }
            return sum;
        }
    
        public static void main(String[] args) {
            ForkJoinPool forkJoinPool = new ForkJoinPool(3);//3个worker的Thread来执行任务
            Future<Integer> future = forkJoinPool.submit(new RecursiveTaskDemo(1, 100));
            System.out.println("处理其他程序...");
            try {
                System.out.println("计算的值为：" + future.get());//get方法阻塞，返回Future的结果
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    
    }

问题：这里的compute方法没有显示的调用自身（递归），怎么将任务不断划分呢？它的子任务会被提交给某个workQueue（双端队列），被对应的ForkJoinPool 中的某个线程按序执行；执行过程中会调用compute方法；

**\*原理分析：**

Fork/Join Framework 的实现算法并不是那么“显然”，而是一个更加复杂的算法——这个算法的名字就叫做 work stealing （工作窃取）算法。

1.  ForkJoinPool的每个工作线程都维护着一个工作队列（WorkQueue），这是一个双端队列（Deque），里面存放的对象是任务（ForkJoinTask）。
2.  每个工作线程在运行中产生新的任务（通常是因为调用了 fork()）时，会放入工作队列的"队尾"，并且工作线程在处理自己的工作队列时，使用的是LIFO方式，也就是说每次从队尾取出任务来执行。
3.  每个工作线程在处理自己的工作队列同时，会尝试从别的线程对应的任务队列中窃取一个任务，窃取的任务是从其他线程的工作队列的队首拿走的，也就是说工作线程在窃取其他工作线程的任务时，使用的是FIFO方式。( 这个"被窃取的任务"也可以是新加入ForkjoinPool的新任务，有时就会看见一种情况，比如我的电脑CPU6核，理论上除了main线程，有5个worker线程去处理ForkjoinPool，但是我放了5个任务到ForkjoinPool，发现只有3个worker线程在执行任务，说明在新加入的任务时，如果他们已经完成了自身队列的任务，他们也会去"抢"这个新的任务 )
4.  在遇到 join()时，如果需要join的任务尚未完成，则会先处理其他任务，并等待其完成。
5.  在既没有自己的任务，也没有可以窃取的任务时，进入休眠。

*  fork() 做的工作只有一件事，既是把任务推入当前工作线程的工作队列里。
 *  join() 的工作则复杂得多，也是 join() 可以使得线程免于被阻塞的原因——不像同名的 Thread.join()。

**join()的JDK8源码分析**：

1.  检查调用 join() 的线程是否是 ForkJoinThread 线程。如果不是（例如 main 线程），则阻塞当前线程，等待任务完成。如果是，则不阻塞。
2.  查看任务的完成状态，如果已经完成，直接返回结果。
3.  如果任务尚未完成，但处于自己的工作队列内，则完成它。
4.  如果任务已经被其他的工作线程偷走，则窃取这个"小偷线程"的工作队列内的任务（以 FIFO 方式），执行，以期帮助它早日完成欲 join 的任务。
5.  如果偷走任务的"小偷线程"也已经把自己的任务全部做完，正在等待需要 join 的任务时，则找到"偷过它任务的线程"，帮助它完成它的任务。
6.  递归地执行第5步。

(图)

![35755fc3bc2a58295a33ecb579d051f9.png][]

**fork & join**

所谓work-stealing模式，即每个工作线程都会有自己的任务队列。当工作线程完成了自己所有的工作后，就会去“偷”别的工作线程的任务。

假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。但是有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。

**submit方法**

其实除了前面介绍过的每个工作线程自己拥有的工作队列以外，ForkJoinPool 自身也拥有工作队列，这些工作队列的作用是用来接收由外部线程（非 ForkJoinThread 线程）提交过来的任务，而这些工作队列被称为 submitting queue 。（如打印的结果里，外部线程为1）

> ...  
> ForkJoinPool-1-worker-3 ...  
> ForkJoinPool-1-worker-3 ...  
> 计算的值为：5050

submit() 和 fork() 其实没有本质区别，只是提交对象变成了 submitting queue 而已（还有一些同步，初始化的操作）。submitting queue 和其他 work queue 一样，是工作线程”窃取“的对象，因此当其中的任务被一个工作线程成功窃取时，就意味着提交的任务真正开始进入执行阶段。

**ForkJoinPool与ThreadPoolExecutor区别**：

1.  ForkJoinPool中的"每个线程都会有一个与之对应的任务队列"，而ThreadPoolExecutor只有一个队列，并根据queue类型不同，细分出各种线程池
2.  ForkJoinPool能够使用数量有限的线程来完成非常多的具有"父子关系"的任务,ThreadPoolExecutor中根本没有什么父子关系任务
3.  ForkJoinPool在使用过程中，会创"建大量的子任务"，会进行"大量的gc"，但是ThreadPoolExecutor不需要，因此单线程（或者任务分配平均）
4.  ForkJoinPool在多任务，且"任务分配不均（计算密集型且不互相依赖）的时候有优势"，但是在单线程或者"任务分配均匀"的情况下，效率没有ThreadPoolExecutor高，毕竟要进行大量gc子任务

**ForkJoinPool在JDK8中的应用：**

stream中应用ForkJoinPool——parallelStream让部分Java代码自动地以并行的方式执行

Arrays.asList("a1", "a2", "b1", "c2", "c1")
      .parallelStream()
      .filter(s -> {
        System.out.format("filter: %s [%s]\n", s, Thread.currentThread().getName());
        return true;
      })
      .map(s -> {
        System.out.format("map: %s [%s]\n", s, Thread.currentThread().getName());
        return s.toUpperCase();
      })
      .sorted((s1, s2) -> {
        System.out.format("sort: %s <> %s [%s]\n", s1, s2, Thread.currentThread().getName());
        return s1.compareTo(s2);
      })
      .forEach(s -> System.out.format("forEach: %s [%s]\n", s, Thread.currentThread().getName()));

ForkJoinPool在多线程情况下，能够实现工作窃取(Work Stealing)，在该线程池的每个线程中会维护一个队列来存放需要被执行的任务。当线程自身队列中的任务都执行完毕后，它会从别的线程中拿到未被执行的任务并帮助它执行。

ThreadPoolExecutor因为它其中的线程并不会关注每个任务之间任务量的差异。当执行任务量最小的任务的线程执行完毕后，它就会处于空闲的状态(Idle)，等待任务量最大的任务执行完毕。

**因此多任务在多线程中分配不均时，ForkJoinPool效率高。**

[35755fc3bc2a58295a33ecb579d051f9.png]: https://img-blog.csdnimg.cn/img_convert/35755fc3bc2a58295a33ecb579d051f9.png