ForkJoin框架

简介

ForkJoin 是JDK1.7的内容，用于并行执行任务！可以提高效率，特别是在大数据量操作时速率会比普通操作更快！

大数据中：MapReduce 核心思想->把大任务拆分为小任务！

ForkJoin特点：工作窃取

实现原理：双端队列！从上面和下面都可以去拿到任务进行执行！

如下图：线程A和B执行任务，B线程执行完了就可以去执行A线程没有执行完的线程。

不好的地方：如果A线程只有一个线程没有执行完成，B线程执行完成，A线程的最后一个任务会造成AB线程抢这个任务去执行。

示例

1. ForkJoinPool的execute方法

2. 点击execute的参数：查看ForkJoinTask类的详情

3. 接下来的示例我们需要返回值，于是选择ForkJoinTask的子类：RecursiveTask

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

/**
 * 如何使用于forkjoin
 * 1. 通过 forkjoinPooL来执行
 * 2. 计算任务 :forkjoinPooL.execute(ForkJoinTask task)
 * 3. 计算类要继承ForkJoinTask
 */

//实现步骤第一步：继承RecursiveTask，指明要返回的值。
public class ForkJoinExample extends RecursiveTask<Long> {
    private static final long SUM = 10_0000_0000;
    private Long start;
    private Long end;
    //临界值
    private Long temp=10000L;

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

    //实现步骤第二步：实现继承类的抽象方法
    @Override
    protected Long compute() {
        //必须写这个判断：通过debug后我们发现join后又会从新执行compute
        if((end-start)<temp){
            Long sum=0L;
            for (Long i = start; i <=end; i++) {
                sum+=i;
            }
            return sum;
        }else {
            //中间值
            long middle=(start+end)/2;
            //将大任务分成两个任务。
            ForkJoinExample task1=new ForkJoinExample(start,middle);
            task1.fork(); //拆分任务，把任务压入线程队列
            ForkJoinExample task2=new ForkJoinExample(middle+1,end);
            task2.fork();
            //获取结果
            return task1.join()+task2.join();
        }
    }

    //测试一使用简单的for循环来
    public static void test01(){
        long star = System.currentTimeMillis();
        long sum = 0L;
        for (Long i = 1L; i <= SUM; i++) {
            sum+=i;
        }
        long end = System.currentTimeMillis();
        System.out.println("累加结果="+sum);
        System.out.println("普通方式执行时间：" + (end-star));
        System.out.println("==================");
    }
    //使用ForkJoin来进行累加操作
    public static void test02() throws ExecutionException, InterruptedException {
        long star = System.currentTimeMillis();

        //实现步骤第三步：使用ForkJoinPool执行
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Long> task = new ForkJoinExample(0L,SUM);
        //forkJoinPool.execute(task); execute 没有返回值
        ForkJoinTask<Long> submit = forkJoinPool.submit(task);
        //使用submit可以获取到结果，但是它有一个阻塞的过程，需要抛出异常
        Long sum = submit.get();
        System.out.println("累加结果="+sum);
        long end = System.currentTimeMillis();
        System.out.println("使用forkjoin执行时间：" + (end - star));
        System.out.println("==================");
    }
    //使用stream流来完成，效率是最高的
    public static void test03() {
        long star = System.currentTimeMillis();
        long sum = LongStream.rangeClosed(0L,10_0000_0000L).parallel().reduce(0,Long::sum);
        System.out.println(sum);
        long end = System.currentTimeMillis();
        System.out.println("使用stream流运行时间：" + (end - star));
    }
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        test01();
        test02();
        test03();
    }
}

运行结果

累加结果=500000000500000000
普通方式执行时间：3749
==================
累加结果=500000000500000000
使用forkjoin执行时间：3956
==================
500000000500000000
使用stream流运行时间：326

从结果上我们可以看到使用stream流速度是最快的。

按理来说应该是forkjoin的速度比普通循环更快，但我电脑上结果却是forkjoin较慢。（此处一脸懵逼，求懂的大佬赐教）

注意：forkjoin方式我们并没有看到调用compute()，但是实际上是调用了的，并且每次调用join方法后又会重新执行compute(递归调用)【通过debug方式运行，可以看到这一过程】

基于上面的问题，我将程序中的起始值，最终值和临界值由包装类型Long改为基本类型long，测试结果如下

累加结果=500000000500000000
普通方式执行时间：488
==================
累加结果=500000000500000000
使用forkjoin执行时间：478
==================
500000000500000000
使用stream流运行时间：373

结果显示将操作数改为基本类型Long-long会快很多，通常执行时间还是for>forkjoin>stream

标签：end,框架,forkjoin,System,Long,线程,执行,ForkJoin
来源： https://www.cnblogs.com/lanxinren/p/14715824.html