当前位置：首页 > news >正文

什么是ForkJoin

news 来源：原创 2024/4/30 17:27:36

1、什么是ForkJoin？

从JDK1.7开始，Java提供Fork/Join框架用于并行执行任务，它的思想就是讲一个大任务分割成若干小任务，最终汇总每个小任务的结果得到这个大任务的结果。

简单来说：就是借助于ForkJoin线程池，我们可以将一个大的任务分割成多个小的任务并行执行并汇总执行结果。在这里看很像我们之前学过的归并排序的逻辑一样。

2、ForkJoin如何解决问题？

**ForkJoin主要分为两步：

第一：任务切分
第二：结果合并

在这里插入图片描述

简单来说，ForkJoin也是一个线程池，而在ForkJoin线程池里面的每个线程都有自己的一个工作序列，当自己队列中的任务都完成以后，会从其他线程的工作队伍中偷一个任务执行，这样就可以充分利用资源。（在这里需要和ThreadPoolExecutor自定义线程池不同，ThreadPoolExecutor中的所有线程都有共用一个工作序列）

3、ForkJoin核心特性：工作窃取

工作窃取成立的前提条件：所用队列的存储模型为双端队列。

工作窃取：ForkJoinPool的各个工作线程都会维护一个各自的任务队列，减少线程之间对于任务的竞争；

每个线程都会先保证将自己队列中的任务执行完，当自己的任务执行完之后，会去看其他线程的任务队列中是否有未处理完的任务，如果有则会帮助其他线程执行；

为了减少在帮助其他线程执行任务时发生竞争，会使用双端队列来存放任务，被窃取的任务只会从队列的头部获取任务，而窃取任务的线程每次都是从队列的尾部获取任务。
在这里插入图片描述

优点：
充分利用线程进行并行计算，并减少了线程间的竞争。

缺点：
需要给每个线程都要开辟一个空间，是在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且消耗了更多的系统资源，比如创建多个线程和多个双端队列。

4、代码实现：

**
在这里插入图片描述

package com.day20221008;

import java.util.concurrent.RecursiveTask;

public class ForkJoinWork extends RecursiveTask<Long> {
    private Long start;
    private Long end;
    private final static Long critical = 1000L;

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

    @Override
    protected Long compute() {
        Long length = end - start;
        if(length <= critical){
            Long sum = 0l;
            for (Long i =0l;i<=end;i++){
                sum += i;
            }
            return sum;
        }else {
            //第一步拆分任务
            Long middle = (start + end)/2;
            ForkJoinWork right = new ForkJoinWork(start, middle);
            right.fork();
            ForkJoinWork left = new ForkJoinWork(middle + 1, end);
            left.fork();

            //合并
            return right.join() + left.join();
        }

    }
}

package com.day20221008;

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

public class ForkJoinWorkImpl {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        //初始化ForkJoinPool池
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        //创建执行的任务
        ForkJoinWork task = new ForkJoinWork(0L, 20000000L);
        //开始计算
        ForkJoinTask<Long> result = forkJoinPool.submit(task);
        Long sum = result.get();
        long end = System.currentTimeMillis();
        System.out.println("result = "+sum+"，总共花费"+(end - start));
    }
}