JUC并发编程--- ForkJoin与Stream并行流

Posted 2021-07-22 小样5411

ForkJoin讲解

ForkJoin是在JDK1.7出来的，在大数据环境下，并行执行任务，提高效率。
原理：用的是分支合并的思想，将大任务拆成多个小任务并行，然后再合并成原来任务

ForkJoin的特点：工作窃取
假设有线程A和线程B同时执行队列中的任务，线程B先执行完，然后线程B不能闲着，就会窃取A，后面没有执行完的来执行，这样就提高效率。就是自己执行完，帮别人执行。

我们来举个求和案例，在IDEA中使用ForkJoin

package com.yx.ForkJoin;

import java.util.concurrent.RecursiveTask;

/**
 *如何使用ForkJoin：1、ForkJoinPool通过它来执行 2、计算任务forkjoinPool.execute(ForkJoinTask task) 3、继承实现类RecursiveTask
 */
public class ForkJoinDemo extends RecursiveTask<Long> {

    private Long start;

    private Long end;
    
    //临界值（可变）
    private Long temp=10000L;

    @Override
    protected Long compute() {
        if ((end-start)<temp){
            //正常计算
            Long sum = 0L;
            for (Long i = start ; i <= end; i++) {
                sum += i;
            }
            return sum;
        }else {
            //forkjoin处理大的数据量
            Long middle = (start+end)/2;//中间值
            ForkJoinDemo task1 = new ForkJoinDemo(start,middle);
            task1.fork();//拆分任务
            ForkJoinDemo task2 = new ForkJoinDemo(middle+1,end);
            task2.fork();//拆分任务
            return task1.join()+task2.join();//合并

        }
    }

    public ForkJoinDemo(Long start, Long end) {
        this.start = start;
        this.end = end;
    }
}

对比1累加到10亿，普通求和，ForkJoin与Stream流三者分别需要多少时间

package com.yx.ForkJoin;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.stream.LongStream;

public class test {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        test3();
    }
    //普通求和,sum=500000000500000000，时间=6641
    public static void test1() {
        Long sum = 0L;
        Long start = System.currentTimeMillis();
        for (Long i = 1L; i <= 10_0000_0000; i++) {
            sum+=i;
        }
        Long end = System.currentTimeMillis();
        System.out.println("sum="+sum+"，时间="+(end-start));
    }

    //forkjoin，sum=500000000500000000，时间=5998
    public static void test2() throws ExecutionException, InterruptedException {
        Long start = System.currentTimeMillis();
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Long> task = new ForkJoinDemo(0L,10_0000_0000L);//相当于RecursiveTask类实现ForkJoinTask接口
        ForkJoinTask<Long> submit = forkJoinPool.submit(task);//提交任务
        Long sum = submit.get();//获取数值
        Long end = System.currentTimeMillis();
        System.out.println("sum="+sum+"，时间="+(end-start));
    }
    //stream并行流(最大程度利用cpu资源),sum=500000000500000000，时间=241
    public static void test3() {
        Long start = System.currentTimeMillis();
        Long sum = LongStream.rangeClosed(0L,10_0000_0000L).parallel().reduce(0,Long::sum);//reduce表示结果,0等价于get(0)将结果拿出来
        Long end = System.currentTimeMillis();
        System.out.println("sum="+sum+"，时间="+(end-start));
    }
}

可以看出大数据情况下Stream并行流是最快的，速度相比快几十倍