线程的并发工具类之forkjoin原理和使用（六）

快来打我* 2022-11-20 04:58 216阅读 0赞

今天分析 forkjoin原理和使用：

1、Fork-Join

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

2、分而治之

同时 forkjoin 在处理某一类问题时非常的有用，哪一类问题？分而治之的问 题。十大计算机经典算法：快速排序、堆排序、归并排序、二分查找、线性查找、 深度优先、广度优先、Dijkstra 、动态规划、朴素贝叶斯分类，有几个属于分 而治之？3 个，快速排序、归并排序、二分查找，还有大数据中 M/R 都是。 分治法的设计思想是：将一个难以直接解决的大问题，分割成一些规模较小 的相同问题，以便各个击破，分而治之。 分治策略是：对于一个规模为 n 的问题，若该问题可以容易地解决（比如说 规模 n 较小）则直接解决，否则将其分解为 k 个规模较小的子问题， 这些子问题 互相独立且与原问题形式相同 ( 子问题相互之间有联系就会变为动态规范算法 ) ， 递归地解这些子问题，然后将各子问题的解合并得到原问题的解。这种算法设计 策略叫做分治法。

3、Fork-Join 原理

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70

4、工作密取

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

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1

5、 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_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2

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

5、Fork/Join 的同步用法和异步用法

参见代如 下，同步执行：

  1. //初始化一个数组
  2. public class MakeArray {
  3.     //数组长度
  4.     public static final int ARRAY_LENGTH  = 4000;
  5.     public final static int THRESHOLD = 47;
  7.     public static int[] makeArray() {
  9.         //new一个随机数发生器
  10.         Random r = new Random();
  11.         int[] result = new int[ARRAY_LENGTH];
  12.         for(int i=0;i<ARRAY_LENGTH;i++){
  13.             //用随机数填充数组
  14.             result[i] =  r.nextInt(ARRAY_LENGTH*3);
  15.         }
  16.         return result;
  18.     }
  19. }

forkjoin 类和main方法:

  1. import java.util.concurrent.ForkJoinPool;
  2. import java.util.concurrent.RecursiveTask;
  4. /**
  5.  * ForkJoin执行累加
  6.  */
  7. public class SumArray {
  8. //继承有返回结果的类
  9.     private static class SumTask extends RecursiveTask<Integer>{
  11.         /*阈值*/
  12.         private final static int THRESHOLD = MakeArray.ARRAY_LENGTH/10;
  13.         private int[] src;
  14.         private int fromIndex;
  15.         private int toIndex;
  17.         public SumTask(int[] src, int fromIndex, int toIndex) {
  18.             this.src = src;
  19.             this.fromIndex = fromIndex;
  20.             this.toIndex = toIndex;
  21.         }
  23.         @Override
  24.         protected Integer compute() {
  25.             /*任务的大小是否合适*/
  26.             if (toIndex - fromIndex < THRESHOLD){
  27. //                System.out.println(" from index = "+fromIndex
  28. //                        +" toIndex="+toIndex);
  29.                 int count = 0;
  30.                 for(int i= fromIndex;i<=toIndex;i++){
  31.               //      SleepTools.ms(1);
  32.                      count = count + src[i];
  33.                 }
  34.                 return count;
  35.             }else{
  36.                 //fromIndex....mid.....toIndex
  37.                 int mid = (fromIndex+toIndex)/2;
  38.                 SumTask left = new SumTask(src,fromIndex,mid);
  39.                 SumTask right = new SumTask(src,mid+1,toIndex);
  40.                 invokeAll(left,right);
  41.                 return left.join()+right.join();
  42.             }
  43.         }
  44.     }
  47.     public static void main(String[] args) {
  48.         int[] src = MakeArray.makeArray();
  49.         /*new出池的实例*/
  50.         ForkJoinPool pool = new ForkJoinPool();
  51.         /*new出Task的实例*/
  52.         SumTask innerFind = new SumTask(src,0,src.length-1);
  54.         long start = System.currentTimeMillis();
  56.         pool.invoke(innerFind);
  57.         //System.out.println("Task is Running.....");
  59.         System.out.println("The count is "+innerFind.join()
  60.                 +" spend time:"+(System.currentTimeMillis()-start)+"ms");
  62.     }
  63. }

执行结果：

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3

异步执行：继承此类 RecursiveAction 没有返回值

  1. /**
  2.  *类说明：遍历指定目录（含子目录）找寻指定类型文件
  3.  */
  4. public class FindDirsFiles extends RecursiveAction {
  6.     private File path;
  8.     public FindDirsFiles(File path) {
  9.         this.path = path;
  10.     }
  12.     @Override
  13.     protected void compute() {
  14.         List<FindDirsFiles> subTasks = new ArrayList<>();
  16.         File[] files = path.listFiles();
  17.         if (files!=null){
  18.             for (File file : files) {
  19.                 if (file.isDirectory()) {
  20.                     // 对每个子目录都新建一个子任务。
  21.                     subTasks.add(new FindDirsFiles(file));
  22.                 } else {
  23.                     // 遇到文件，检查。
  24.                     if (file.getAbsolutePath().endsWith("txt")){
  25.                         System.out.println("文件:" + file.getAbsolutePath());
  26.                     }
  27.                 }
  28.             }
  29.             if (!subTasks.isEmpty()) {
  30.                 // 在当前的 ForkJoinPool 上调度所有的子任务。
  31.                 for (FindDirsFiles subTask : invokeAll(subTasks)) {
  32.                     subTask.join();
  33.                 }
  34.             }
  35.         }
  36.     }
  38.     public static void main(String [] args){
  39.         try {
  40.             // 用一个 ForkJoinPool 实例调度总任务
  41.             ForkJoinPool pool = new ForkJoinPool();
  42.             FindDirsFiles task = new FindDirsFiles(new File("E:/"));
  44.             /*异步提交*/
  45.             pool.execute(task);
  47.             /*主线程做自己的业务工作*/
  48.             System.out.println("Task is Running......");
  49.             Thread.sleep(10000);
  50.             int otherWork = 0;
  51.             for(int i=0;i<100;i++){
  52.                 otherWork = otherWork+i;
  53.             }
  54.             System.out.println("Main Thread done sth......,otherWork="
  55.                     +otherWork);
  56.         //    task.join();//阻塞方法 如果执行此方法，则任务执行完 才继续往下走
  57.             System.out.println("Task end");
  58.         } catch (Exception e) {
  59.             // TODO Auto-generated catch block
  60.             e.printStackTrace();
  61.         }
  62.     }
  63. }

执行结果：

20201107164428853.png

6、forkjoin 实现一个归并排序算法：

  1. import java.util.Arrays;
  2. import java.util.concurrent.ForkJoinPool;
  3. import java.util.concurrent.RecursiveTask;
  5. /**
  6.  * forkjoin实现的归并排序
  7.  */
  8. public class FkSort {
  9.     private static class SumTask extends RecursiveTask<int[]>{
  11.         private final static int THRESHOLD = 2;
  12.         private int[] src;
  14.         public SumTask(int[] src) {
  15.             this.src = src;
  16.         }
  18.         @Override
  19.         protected int[] compute() {
  20.             if(src.length<=THRESHOLD){
  21.                 return InsertionSort.sort(src);
  22.             }else{
  23.                 //fromIndex....mid.....toIndex
  24.                 int mid = src.length / 2;
  25.                 SumTask leftTask = new SumTask(Arrays.copyOfRange(src, 0, mid));
  26.                 SumTask rightTask = new SumTask(Arrays.copyOfRange(src, mid, src.length));
  27.                 invokeAll(leftTask,rightTask);
  28.                 int[] leftResult = leftTask.join();
  29.                 int[] rightResult = rightTask.join();
  30.                 return SumTask.merge(leftResult,rightResult);
  31.             }
  32.         }
  34.         /**
  35.          * 归并排序——将两段排序好的数组结合成一个排序数组
  36.          *
  37.          * @param left
  38.          * @param right
  39.          * @return
  40.          */
  41.         public static int[] merge(int[] left, int[] right) {
  42.             int[] result = new int[left.length + right.length];
  43.             for (int index = 0, i = 0, j = 0; index < result.length; index++) {
  44.                 if (i >= left.length)/*左边数组已经取完，完全取右边数组的值即可*/
  45.                     result[index] = right[j++];
  46.                 else if (j >= right.length)/*右边数组已经取完，完全取左边数组的值即可*/
  47.                     result[index] = left[i++];
  48.                 else if (left[i] > right[j])/*左边数组的元素值大于右边数组，取右边数组的值*/
  49.                     result[index] = right[j++];
  50.                 else/*右边数组的元素值大于左边数组，取左边数组的值*/
  51.                     result[index] = left[i++];
  52.             }
  54.             return result;
  55.         }
  57.     }
  60.     public static void main(String[] args) {
  62.         ForkJoinPool pool = new ForkJoinPool();
  63.         int[] src = MakeArray.makeArray();
  65.         SumTask innerFind = new SumTask(src);
  67.         long start = System.currentTimeMillis();
  69.         int[] invoke = pool.invoke(innerFind);
  70.         for(int number:invoke){
  71.             System.out.println(number);
  72.         }
  73.         System.out.println(" spend time:"+(System.currentTimeMillis()-start)+"ms");
  75.     }
  76. }

执行结果：

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4

到此forkjoin 解析完了。

发表评论

表情：
评论列表 （有 0 条评论，216人围观）

还没有评论，来说两句吧...

相关阅读

    相关 线程的并发工具类

    Fork-Join 什么是分而治之？ 规模为N的问题，N<阈值，直接解决，N>阈值，将N分解为K个小规模子问题，子问题互相对立，与原问题形式相同，将子问题的解合并得到

    蔚落蔚落/ 0 赞/ 396 阅读