Fork/Join——分治编程

客官°小女子只卖身不卖艺 2021-08-30 12:46 265阅读 0赞

### Fork/Join ###

在JDK1.7版本中提供了Fork/Join 并行执行任务框架，它的主要作用是把大任务分割成若干个小任务，再对每个小任务得到的结果进行汇总，此种开发方法也叫分治编程，分治编程可以极大地利用CPU资源，提高任务执行的效率，也是目前与多线程有关的前沿技术。

#### Fork-Join分治编程与类结构 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

在JDK中并行执行框架Fork-Join使用了`“工作窃取( work- stealing)”`算法，它是指某个线程从其他队列里窃取任务来执行，那这样做有什么优势或者目的是什么呢?

比如要完成一个 比较大的任务，完全可以把这个大的任务分割为若千互不依赖的子任务/小任务，为了更加方便地管理这些任务，于是把这些子任务分别放到不同的队列里，这时就会出现有的线程会先把自己队列里的任务快速执行完毕，而其他线程对应的队列里还有任务等待处理，完成任务的线程与其等着，不如去帮助其他线程分担要执行的任务，于是它就去其他线程的队列里窃取一个任务来执行，这就是所谓的“工作窃取(work-stealing)” 算法。

在JDK.7中实现中实现分治编程需要使用ForkJoinPool类，此类的主要作用是创建一个任务池，类信息如下：

public class ForkJoinPoll extends AbstractExecutorService{ }

该类也是从AbstractExxcutorService类继承下来的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

类ForkJoinPool所提供的功能是一个任务池，而执行任务却不是ForkJoinPool，而是ForkJoinTask类。

类FokJoinTask是抽象类，不能实例化，所以需要该类的3个子类`CountedCompleter、RecursiveAction和RecursiveTask`来实现具体功能。

#### 使用RecursiveAction让任务跑起来 ####

使用类RecursiveAction执行的任务是具有`无返回值的，仅执行一次任务`。

public class MyTestA { 
        public static void main(String[] args) throws InterruptedException { 
            ForkJoinPool poll = new ForkJoinPool();
            poll.submit(new MyRecursiveAction());
            Thread.sleep(5000);
        }
    }
    
    class MyRecursiveAction extends RecursiveAction { 
    
        @Override
        protected void compute() { 
            System.out.println("compute run!");
        }
    }

![在这里插入图片描述][20200625104913895.png]

#### 使用RecursiveAction分解任务 ####

前面的示例仅是让任务运行起来，并打印一个字符串信息，任务并没有得到fork分解，也就是并没有体现分治编程的运行效果。在调用ForkJoinTask.java类中的fork() 方法时需要注意一下效率的问题，因为每一次调用fork 都会分离任务，增加系统运行负担，所以在ForkJoinTask.java类中提供了`public static void invokeAll(ForkJoinTask<?> tl, ForkJoinTask<?>t2)`方法来优化执行效率。

public class MyTestA { 
        public static void main(String[] args) throws InterruptedException { 
            ForkJoinPool poll = new ForkJoinPool();
            poll.submit(new MyRecursiveAction(1, 10));
            Thread.sleep(5000);
        }
    }
    
    class MyRecursiveAction extends RecursiveAction { 
        private int beginValue;
        private int endValue;
    
        public MyRecursiveAction(int beginValue, int endValue) { 
            this.beginValue = beginValue;
            this.endValue = endValue;
        }
    
        @Override
        protected void compute() { 
            System.out.println(Thread.currentThread().getName() + "------");
            if (endValue - beginValue > 2) { 
                int middleNum = (beginValue + endValue) / 2;
                MyRecursiveAction leftAction = new MyRecursiveAction(beginValue, middleNum);
                MyRecursiveAction rightAction = new MyRecursiveAction(middleNum + 1, endValue);
                invokeAll(leftAction, rightAction);
            } else { 
                System.out.println("打印组合为：" + beginValue + "-" + endValue);
            }
        }
    }

任务被成功分解。

ForkJoinPool-1-worker-9------
    ForkJoinPool-1-worker-9------
    ForkJoinPool-1-worker-9------
    打印组合为：1-3
    ForkJoinPool-1-worker-2------
    ForkJoinPool-1-worker-11------
    ForkJoinPool-1-worker-4------
    ForkJoinPool-1-worker-2------
    打印组合为：4-5
    打印组合为：6-8
    打印组合为：9-10

#### 使用RecursiveTask取得返回值与join()和get()方法的区别 ####

使用类RecursiveTask执行的任务具有返回值的功能。

public class MyTestB { 
        public static void main(String[] args) { 
            try { 
                MyResursiveTask task1 = new MyResursiveTask();
                System.out.println(task1.hashCode());
                ForkJoinPool pool = new ForkJoinPool();
                ForkJoinTask task2 = pool.submit(task1);
                System.out.println(task2.hashCode() + " " + task2.get());
                Thread.sleep(5000);
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }
    
    class MyResursiveTask extends RecursiveTask<Integer> { 
    
        @Override
        protected Integer compute() { 
            System.out.println("compute time=" + System.currentTimeMillis());
            return 100;
        }
    }

![在这里插入图片描述][20200625110454171.png]

可以使用join()方法来取得结果值：

public static void main(String[] args) { 
            try { 
                MyResursiveTask task1 = new MyResursiveTask();
                System.out.println(task1.hashCode());
                ForkJoinPool pool = new ForkJoinPool();
                ForkJoinTask<Integer> task2 = pool.submit(task1);
                System.out.println(task2.hashCode() + " " + task2.join());
                Thread.sleep(5000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }

![在这里插入图片描述][20200625110747673.png]

方法join()与get()虽然都能取得计算后的结果值，但它们之间还是在出现异常时有处理上的区别：

public class MyTestC { 
        public static void main(String[] args) { 
            try { 
                MyRecursiveTaskA taskA = new MyRecursiveTaskA();
                ForkJoinPool pool = new ForkJoinPool();
                ForkJoinTask<Integer> task = pool.submit(taskA);
                System.out.println(task.get());
                for (int i = 0; i < Integer.MAX_VALUE; i++) { 
                    String newString = new String();
                    Math.random();
                    Math.random();
                    Math.random();
                    Math.random();
                    Math.random();
                    Math.random();
                    Math.random();
                }
            } catch (Exception e) { 
                e.printStackTrace();
            }
            System.out.println("main end");
        }
    }
    
    class MyRecursiveTaskA extends RecursiveTask<Integer> { 
    
        @Override
        protected Integer compute() { 
            System.out.println(Thread.currentThread().getName() + " 执行了compute方法");
            String nullString = null;
            nullString.toString();//NullPoint
            return 100;
        }
    }

使用get()方法执行任务时，当子任务出现异常时可以在main主线程中进行捕获。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]

#### 使用RecursiveTask执行多个任务并打印返回值 ####

public class MyTestD { 
        public static void main(String[] args) throws InterruptedException { 
            ForkJoinPool pool = new ForkJoinPool();
            ForkJoinTask<Integer> taskA = pool.submit(new AMyRecursiveTask());
            ForkJoinTask<Integer> taskB = pool.submit(new BMyRecursiveTask());
            System.out.println("准备打印：" + System.currentTimeMillis());
            System.out.println(taskA.join() + " A: " + System.currentTimeMillis());
            System.out.println(taskB.join() + " B: " + System.currentTimeMillis());
            Thread.sleep(5000);
        }
    }
    class AMyRecursiveTask extends RecursiveTask<Integer> { 
    
        @Override
        protected Integer compute() { 
            try { 
                System.out.println(Thread.currentThread().getName() + " begin A" + System.currentTimeMillis());
                Thread.sleep(3000);
                System.out.println(Thread.currentThread().getName() + " end B" + System.currentTimeMillis());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            return 100;
        }
    }
    class BMyRecursiveTask extends RecursiveTask<Integer> { 
    
        @Override
        protected Integer compute() { 
            try { 
                System.out.println(Thread.currentThread().getName() + " begin B" + System.currentTimeMillis());
                Thread.sleep(5000);
                System.out.println(Thread.currentThread().getName() + " end B" + System.currentTimeMillis());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            return 100;
        }
    }

每个任务成功返回100，并且任务之间运行的方式是异步的，但join()方法却是同步的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]

#### RecursiveTask实现字符串累加 ####

public class MyTestE { 
        public static void main(String[] args) throws InterruptedException { 
            ForkJoinPool pool = new ForkJoinPool();
            MyRecursiveTaskForString task = new MyRecursiveTaskForString(1, 20);
            ForkJoinTask<String> task1 = pool.submit(task);
            System.out.println(task1.join());
            Thread.sleep(5000);
        }
    }
    
    class MyRecursiveTaskForString extends RecursiveTask<String> { 
        private int beginValue;
        private int endValue;
    
        public MyRecursiveTaskForString(int beginValue, int endValue) { 
            this.beginValue = beginValue;
            this.endValue = endValue;
        }
    
        @Override
        protected String compute() { 
            System.out.println(Thread.currentThread().getName() + " ------------");
            if (endValue - beginValue > 2) { 
                int middleValue = (endValue + beginValue) / 2;
                MyRecursiveTaskForString leftTask = new MyRecursiveTaskForString(beginValue, middleValue);
                MyRecursiveTaskForString rightTask = new MyRecursiveTaskForString(middleValue + 1, endValue);
                invokeAll(leftTask, rightTask);
                return leftTask.join() + rightTask.join();
            } else { 
                String returnString = "";
                for (int i = beginValue; i <= endValue;i++) { 
                    returnString = returnString +  (i);
                }
                System.out.println("else返回：" + returnString + " " + beginValue + " " + endValue);
                return returnString;
            }
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]

#### 使用Fork-Join实现求和：示例1 ####

public class MyTestF { 
        public static void main(String[] args) { 
            try { 
                MyRecursiveTask task = new MyRecursiveTask(1, 10);
                ForkJoinPool pool = new ForkJoinPool();
                pool.submit(task);
                System.out.println("结果值为：" + task.get());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (ExecutionException e) { 
                e.printStackTrace();
            }
        }
    }
    
    class MyRecursiveTask extends RecursiveTask<Integer> { 
        private int beginPosition;
        private int endPosition;
    
        public MyRecursiveTask(int beginPosition, int endPosition) { 
            this.beginPosition = beginPosition;
            this.endPosition = endPosition;
            System.out.println("# " + (beginPosition + " " + endPosition));
        }
    
        @Override
        protected Integer compute() { 
            System.out.println(Thread.currentThread().getName() + "--------");
            Integer sumValue = 0;
            System.out.println("compute=" + beginPosition + " " + endPosition);
            if ((endPosition - beginPosition) != 0) { 
                System.out.println("!=0");
                int middleNum = (endPosition + beginPosition) / 2;
                System.out.println("left 传入的值：" + (beginPosition + " " + middleNum));
                MyRecursiveTask leftTask = new MyRecursiveTask(beginPosition, middleNum);
                System.out.println("right 传入的值：" + (middleNum + 1) + " " + endPosition);
                MyRecursiveTask rightTask = new MyRecursiveTask(middleNum + 1, endPosition);
    
                invokeAll(leftTask, rightTask);
    
                Integer left = leftTask.join();
                Integer right = rightTask.join();
    
                return left + right;
            } else { 
                return endPosition;
            }
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]

#### 使用Fork-Join实现求和：示例2 ####

public class MyTestG { 
        public static void main(String[] args) { 
            MyRecursiveTaskSum task = new MyRecursiveTaskSum(1, 10);
            ForkJoinPool pool = new ForkJoinPool();
            ForkJoinTask<Integer> submit = pool.submit(task);
            System.out.println("结果值为：" + submit.join());
        }
    }
    
    class MyRecursiveTaskSum extends RecursiveTask<Integer> { 
        private int beginPostion;
        private int endPosition;
    
        public MyRecursiveTaskSum(int beginPostion, int endPosition) { 
            this.beginPostion = beginPostion;
            this.endPosition = endPosition;
            System.out.println("# " + (beginPostion + " " + endPosition));
        }
    
        @Override
        protected Integer compute() { 
            Integer sumValue = 0;
            System.out.println("compute=" + beginPostion + " " + endPosition);
            if ((endPosition - beginPostion) > 2) { 
                System.out.println("!=0");
                int middleNum = (endPosition + beginPostion) / 2;
                System.out.println("left 传入的值：" + (beginPostion + " " + middleNum));
                MyRecursiveTaskSum leftTask = new MyRecursiveTaskSum(beginPostion, middleNum);
                System.out.println("right 传入的值：" + (middleNum + 1) + " " + endPosition);
                MyRecursiveTaskSum rightTask = new MyRecursiveTaskSum(middleNum + 1, endPosition);
                invokeAll(leftTask, rightTask);
    
                Integer leftValue = leftTask.join();
                Integer rightValue = rightTask.join();
                System.out.println("++++++++++++++++++" + (leftValue + rightValue));
                return leftValue + rightValue;
            } else { 
                int count = 0;
                for (int i = beginPostion; i <= endPosition; i++) { 
                    count += i;
                }
                return count;
            }
        }
    }

核心条件为`if((endPosition-beginPosition)>2)`，也就是可以使用else\{\}中的for循环以数字范围的方式进行累加求和。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 8]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/39b4a4d54f6242dea8930b7cb5e2d59a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/e0a2faf2bd68411e913c41ec1bd49793.png
[20200625104913895.png]: /images/20210813/e10ebc127d69487085108fd5ca997fe7.png
[20200625110454171.png]: /images/20210813/13537c2a37e64e799b886bb3ef0db718.png
[20200625110747673.png]: /images/20210813/e9468a8096d44aaba47f4e6a36ce77ec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/8115fb34248242eb82e4e4347c118935.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/9dd83bc2aa3f4870b1c0f4fa6a5a9142.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/01f0db144f5c448eb4a4820fc650bed2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]: /images/20210813/a28cd6e06beb4855bb7480f991a70039.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]: /images/20210813/6d2bbd4b73f44bca92649555fe0ea7f4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]: /images/20210813/997d62b210a946b68b0f15d861af11a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 8]: /images/20210813/e32c5bf808ab4f47bc1d13bc5edee7b1.png