Java ForkJoinPool分支合并框架

亦凉 2023-02-20 10:56 117阅读 0赞

### 一、ForkJoinPool 分支合并框架 ###

ForkJoinPool类实现了ExecutorService接口，因此也属于线程池，是一种特殊的线程池。

ForkJoinPool这个工具类从Java7 才开始提供的，优势在于，可以充分利用多cpu，多核cpu的优势，把一个大任务(fork)成若干小任务，把若干小任务放到多个处理器核心上并行执行；当这些小任务执行完成之后，最终汇总(Join)每个小任务结果后得到大任务结果的框架。从而实现用少量的线程，完成大数量的任务。

**1、ThreadPoolExecutor 与 ForkJoinPool区别**

ForkJoinPool与ThreadPoolExecutor两个都实现了Executor和ExecutorService接口。

ForkJoinPool它使用了一个无限队列来保存需要执行的任务，而线程的数量则是通过构造函数传入，如果没有向构造函数中传入希望的线程数量，那么当前计算机可用的CPU数量会被设置为线程数量作为默认值。使用ForkJoinPool能够使用数量有限的线程来完成非常多的具有父子关系的任务，

ThreadPoolExecutor能够提高线程的可管理性，通过重用已存在的线程，降低线程创建和销毁造成的消耗；提高系统响应速度。但是，由于ThreadPoolExecutor中的Thread无法选择优先执行子任务，所以不能完成父子关系的任务。

**2、使用方法**

创建 **ForkJoinPool** 实例，然后调用ForkJoinPool的 **submit**(ForkJoinTask<T> task) 或**invoke**(ForkJoinTask<T> task)方法来执行指定任务了。其中 ForkJoinTask代表一个可以并行、合并的任务。

**ForkJoinTask**是一个抽象类，它还有两个抽象子类：RecusiveAction和RecusiveTask。其中**RecusiveTask代表有返回值的任务**，而**RecusiveAction代表没有返回值的任务。**

*   *  <table> 
         <tbody> 
          <tr> 
           <td><code>&lt;T&gt;&nbsp;<a rel="nofollow">ForkJoinTask</a>&lt;T&gt;</code></td> 
           <td><code><a rel="nofollow">submit</a>(<a rel="nofollow">ForkJoinTask</a>&lt;T&gt;&nbsp;task)</code> <p>提交forkjointask执行。</p> </td> 
          </tr> 
          <tr> 
           <td><code><a rel="nofollow">ForkJoinTask</a>&lt;?&gt;</code></td> 
           <td><code><a rel="nofollow">submit</a>(<a rel="nofollow">Runnable</a>&nbsp;task)</code> <p>提交执行一个Runnable任务并返回一个表示该任务的未来。</p> </td> 
          </tr> 
          <tr> 
           <td><code>&lt;T&gt;&nbsp;<a rel="nofollow">ForkJoinTask</a>&lt;T&gt;</code></td> 
           <td><code><a rel="nofollow">submit</a>(<a rel="nofollow">Runnable</a>&nbsp;task, T&nbsp;result)</code> <p>提交执行一个Runnable任务并返回一个表示该任务的未来。</p> </td> 
          </tr> 
          <tr> 
           <td><code>&lt;T&gt;&nbsp;T</code></td> 
           <td><code><a rel="nofollow">invoke</a>(<a rel="nofollow">ForkJoinTask</a>&lt;T&gt;&nbsp;task)</code> <p>完成给定的任务，完成后返回其结果。</p> </td> 
          </tr> 
          <tr> 
           <td><code>&lt;T&gt;&nbsp;<a rel="nofollow">List</a>&lt;<a rel="nofollow">Future</a>&lt;T&gt;&gt;</code></td> 
           <td><code><a rel="nofollow">invokeAll</a>(<a rel="nofollow">Collection</a>&lt;? extends <a rel="nofollow">Callable</a>&lt;T&gt;&gt;&nbsp;tasks)</code> <p>执行给定的任务，返回一个未来持有他们的状态和结果的列表时，所有的完整。</p> </td> 
          </tr> 
          <tr> 
           <td><code>boolean</code></td> 
           <td><code><a rel="nofollow">awaitTermination</a>(long&nbsp;timeout, <a rel="nofollow">TimeUnit</a>&nbsp;unit)</code> <p>直到所有的任务都完成后，关闭请求，或超时发生，或当前线程被中断，以先发生的情况。</p> <p>&nbsp;</p> </td> 
          </tr> 
          <tr> 
           <td><code>static <a rel="nofollow">ForkJoinPool</a></code></td> 
           <td><code><a rel="nofollow">commonPool</a>()</code> <p>返回公用池实例。</p> </td> 
          </tr> 
         </tbody> 
        </table>

**实例一：没有返回值的“大任务”(打印1-3000之间的整数）- RecusiveAction**

public class ForkJoinPoolDemo1 {
    
        public static void main(String[] args) throws Exception {
            PrintTask task = new PrintTask(1, 300);
            // 创建ForkJoinPool实例，并执行分割任务，
            // 若参数为空，则当前计算机可用的CPU数量会被设置为线程数量作为默认值
            ForkJoinPool pool = new ForkJoinPool(4);
            pool.submit(task);
            // 线程阻塞，等待所有任务完成
            pool.awaitTermination(2, TimeUnit.SECONDS);
            // 关闭线程池
            pool.shutdown();
        }
    }
    
    class PrintTask extends RecursiveAction {
        private static final int THRESHOLD = 50; //一次最多只能打印50个数
        private int start;
        private int end;
    
        public PrintTask(int start, int end) {
            this.start = start;
            this.end = end;
        }
    
        @Override
        protected void compute() {
            if (end - start < THRESHOLD) {
                for (int i = start; i < end; i++) {
                    System.out.println(Thread.currentThread().getName() + "的i值：" + i);
                }
            } else {
                // 把大任务对半拆分成小任务，使用递归
                int middle = (start + end) / 2;
                PrintTask left = new PrintTask(start, middle);
                PrintTask right = new PrintTask(middle, end);
                //并行执行两个小任务
                left.fork();
                right.fork();
            }
        }
    }

**实例二：有返回值的“大任务”(对一个长度为300的数组元素进行累加求和）- RecusiveTask**

public class ForkJoinPoolDemo2 {
        public static void main(String[] args) throws Exception {
            int[] arr = new int[300];
            Random random = new Random();
            int total = 0;
            // 初始化100个数组元素
            for (int i = 0, len = arr.length; i < len; i++) {
                int temp = random.nextInt(20);
                // 对数组元素赋值，并将数组元素的值添加到sum总和中
                total += (arr[i] = temp);
            }
            System.out.println("初始化数组总和：" + total);
    
            SumTask task = new SumTask(arr, 0, arr.length);
            // 创建ForkJoinPool，这个是jdk1.8提供的功能
            ForkJoinPool pool = ForkJoinPool.commonPool();
            Future<Integer> future = pool.submit(task); //提交分解的SumTask 任务
            System.out.println("多线程执行结果：" + future.get());
            //关闭线程池
            pool.shutdown();
        }
    }
    
    class SumTask extends RecursiveTask<Integer> {
        private static final int THRESHOLD = 20; //每个小任务 最多只累加20个数
        private int arry[];
        private int start;
        private int end;
    
        public SumTask(int[] arry, int start, int end) {
            super();
            this.arry = arry;
            this.start = start;
            this.end = end;
        }
    
        @Override
        protected Integer compute() {
            int sum = 0;
            //当end与start之间的差小于threshold时，开始进行实际的累加
            if (end - start < THRESHOLD) {
                for (int i = start; i < end; i++) {
                    sum += arry[i];
                }
                return sum;
            } else {
                //当end与start之间的差大于threshold，即要累加的数超过20个时候，将大任务分解成小任务
                int middle = (start + end) / 2;
                SumTask left = new SumTask(arry, start, middle);
                SumTask right = new SumTask(arry, middle, end);
                //并行执行两个小任务
                left.fork();
                right.fork();
                //把两个小任务累加的结果合并起来
                return left.join() + right.join();
            }
        }
    }

![20200626154722965.png][]

参考文章：

[ForkJoinPool 分支/ 合并框架实战与原理分析][ForkJoinPool _]

—— Stay Hungry. Stay Foolish. 求知若饥，虚心若愚。

[20200626154722965.png]: https://img-blog.csdnimg.cn/20200626154722965.png
[ForkJoinPool _]: https://blog.csdn.net/JSUTDOIT/article/details/81114144