ForkJoin介绍跟使用

快来打我* 2021-10-18 13:36 378阅读 0赞

## Fork-Join  ##

java下多线程的开发可以我们自己启用多线程，线程池，还可以使用forkjoin，forkjoin可以让我们不去了解诸如Thread,Runnable等相关的知识，只要遵循forkjoin的开发模式，就可以写出很好的多线程并发程序，

### 分而治之的思想 ###

同时forkjoin在处理某一类问题时非常的有用，哪一类问题？分而治之的问题。十大计算机经典算法：快速排序、堆排序、归并排序、二分查找、线性查找、

深度优先、广度优先、Dijkstra、动态规划、朴素贝叶斯分类，有几个属于分而治之？3个，快速排序、归并排序、二分查找，还有大数据中M/R都是。

分治法的设计思想是：将一个难以直接解决的大问题，分割成一些规模较小的相同问题，以便各个击破，分而治之。

分治策略是：对于一个规模为n的问题，若该问题可以容易地解决（比如说规模n较小）则直接解决，否则将其分解为k个规模较小的子问题，**这些子问题互相独立且与原问题形式相同****(****子问题相互之间有联系就会变为动态规范算法****)**，递归地解这些子问题，然后将各子问题的解合并得到原问题的解。这种算法设计策略叫做分治法。

### Fork-Join原理 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NzMwMzM4_size_16_color_FFFFFF_t_70][]

*工作密取*

即当前线程的Task已经全被执行完毕，则自动取到其他线程的Task池中取出Task继续执行。

ForkJoinPool中维护着多个线程（一般为CPU核数）在不断地执行Task，每个线程除了执行自己职务内的Task之外，还会根据自己工作线程的闲置情况去获取其他繁忙的工作线程的Task，如此一来就能能够减少线程阻塞或是闲置的时间，提高CPU利用率。

![20190812124501786.png][]

### Fork/Join实战 ###

*Fork/Join使用的标准范式*

我们要使用ForkJoin框架，必须首先创建一个ForkJoin任务。它提供在任务中执行fork和join的操作机制，通常我们不直接继承ForkjoinTask类，只需要直接继承其子类。

1. RecursiveAction，用于没有返回结果的任务

2. RecursiveTask，用于有返回值的任务

task要通过ForkJoinPool来执行，使用submit 或 invoke 提交，两者的区别是：invoke是同步执行，调用之后需要等待任务完成，才能执行后面的代码；submit是异步执行。

join()和get方法当任务完成的时候返回计算结果。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NzMwMzM4_size_16_color_FFFFFF_t_70 1][]

在我们自己实现的compute方法里，首先需要判断任务是否足够小，如果足够小就直接执行任务。如果不足够小，就必须分割成两个子任务，每个子任务在调用invokeAll方法时，又会进入compute方法，看看当前子任务是否需要继续分割成孙任务，如果不需要继续分割，则执行当前子任务并返回结果。使用join方法会等待子任务执行完并得到其结果。

Fork/Join例子

1. 用RecursiveTask

需求：要求对4000万个数进行累加求和

第一步随机产生400万个数

package cn.enjoyedu.ch2.forkjoin.sum;
    
    import java.util.Random;
    
    public class MakeArray {
        //数组长度
        public static final int ARRAY_LENGTH  = 40000000;
       //阈值的设置
        public final static int THRESHOLD = 47;
    
        public static int[] makeArray() {
    
            //new一个随机数发生器
            Random r = new Random();
            int[] result = new int[ARRAY_LENGTH];
            for(int i=0;i<ARRAY_LENGTH;i++){
                //用随机数填充数组
                result[i] =  r.nextInt(ARRAY_LENGTH*3);
            }
            return result;
    
        }
    }

第二步 写任务类

package cn.enjoyedu.ch2.forkjoin.sum;
    
    import cn.enjoyedu.tools.SleepTools;
    
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.RecursiveTask;
    
    /**
     * ForkJoin执行累加
     */
    public class SumArray {
        private static class SumTask extends RecursiveTask<Integer>{
    
            /*阈值*/
            private final static int THRESHOLD = MakeArray.ARRAY_LENGTH/10;
            private int[] src;
            private int fromIndex;
            private int toIndex;
    
            public SumTask(int[] src, int fromIndex, int toIndex) {
                this.src = src;
                this.fromIndex = fromIndex;
                this.toIndex = toIndex;
            }
    
            @Override
            protected Integer compute() {
                /*任务的大小是否合适*/
                //符合你设定的规则
                if (toIndex - fromIndex < THRESHOLD){
                    
                    //实际业务内容此处将数组分成47/10 5条纪录内时，进行求和，然后再合并
                    
    //                System.out.println(" from index = "+fromIndex
    //                        +" toIndex="+toIndex);
                    int count = 0;
                    for(int i= fromIndex;i<=toIndex;i++){
                    	SleepTools.ms(1);
                         count = count + src[i];
                    }
                    return count;
                }else{
                //当不符合的时候继续拆分直到符合
                    //fromIndex....mid.....toIndex
                    int mid = (fromIndex+toIndex)/2;
                    SumTask left = new SumTask(src,fromIndex,mid);
                    SumTask right = new SumTask(src,mid+1,toIndex);
                    invokeAll(left,right);
                    return left.join()+right.join();
                }
            }
        }
    
    
        public static void main(String[] args) {
    
            int[] src = MakeArray.makeArray();
            /*new出池的实例*/
            ForkJoinPool pool = new ForkJoinPool();
            /*new出Task的实例*/
            SumTask innerFind = new SumTask(src,0,src.length-1);
    
            long start = System.currentTimeMillis();
    
            pool.invoke(innerFind);
            //System.out.println("Task is Running.....");
            //innerFind.join()  得到最终求和数据
            System.out.println("The count is "+innerFind.join()
                    +" spend time:"+(System.currentTimeMillis()-start)+"ms");
    
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NzMwMzM4_size_16_color_FFFFFF_t_70]: /images/20211018/abdc92d1b2744a5387804dc911ba6228.png
[20190812124501786.png]: /images/20211018/4174f075f8f34b22bacdbc0b7c0a5ab1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NzMwMzM4_size_16_color_FFFFFF_t_70 1]: /images/20211018/96104553634448ca8b1f3c8e0d755d2b.png