Java多线程任务分解-ForkJoinPool【带返回值】

迈不过友情╰ 2022-09-15 00:08 135阅读 0赞

# 1、什么是ForkJoinPool #

Fork/Join框架，是⼀个实现了ExecutorService接⼝的多线程处理器，它专为那些可以通过递归分解成更细⼩的任务⽽设计，最⼤化的利⽤多核处理器来提⾼应⽤程序的性能。

与其他ExecutorService相关的实现相同的是，Fork/Join框架会将任务分配给线程池中的线程。⽽与之不同的是，Fork/Join框架在执⾏任务时使⽤了⼯作窃取算法。

fork在英⽂⾥有分叉的意思，join在英⽂⾥连接、结合的意思。顾名思义，fork就是要使⼀个⼤任务分解成若⼲个⼩任务，⽽join就是最后将各个⼩任务的结果结合起来得到⼤任务的结果。

Fork/Join的运⾏流程⼤致如下所示：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5pil6aOO5YyW5L2c56eL6Zuo_size_10_color_FFFFFF_t_70_g_se_x_16][]

#  2、多线程任务分解实现 #

package threads;
    
    import java.util.concurrent.RecursiveTask;
    
    /**
     * 继承RecursiveTask来实现支持返回值的"可分解"的任务
     */
    class CalculateTask extends RecursiveTask<Integer> {
        //每个任务处理上限
        private static final int THRESHOLD = 20;
        private int arr[];
        private int start;
        private int end;
    
        public CalculateTask(int[] arr, int start, int end) {
            this.arr = arr;
            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 += arr[i];
                }
                return sum;
            }
            //当end与start之间的差大于THRESHOLD时，将任务分解成两个小任务。
            int middle = (start + end) / 2;
            CalculateTask left = new CalculateTask(arr, start, middle);
            CalculateTask right = new CalculateTask(arr, middle, end);
    
            //并行执行两个“小任务”
            left.fork();
            right.fork();
    
            //把两个“小任务”累加的结果合并起来
            return left.join() + right.join();
    
        }
    }

# 3、多线程任务分解测试 #

package threads;
    
    import java.util.Date;
    import java.util.Random;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.Future;
    
    public class CalculateTaskTest {
    
        public static void main(String[] args) throws Exception {
            int[] arr = new int[10000000];
            Random rand = new Random();
            int total = 0;
            for (int i = 0, len = arr.length; i < len; i++) {
                arr[i] = rand.nextInt(20);
            }
    
            System.out.println("--------------------------------------");
            System.out.println("单线程循环遍历累计加和，任务开始...");
            Long now = new Date().getTime();
            for (int i = 0, len = arr.length; i < len; i++) {
                total += arr[i];
            }
            Long consume = new Date().getTime() - now;
            System.out.println("单线程循环遍历累计加和，任务结束;耗时：" + consume);
    
            System.out.println("单线程循环遍历累计加和：" + total);
            System.out.println("--------------------------------------");
            System.out.println("多线程任务分解就算加和，任务开始...");
            now = new Date().getTime();
    
            // 创建通用池
            ForkJoinPool pool = ForkJoinPool.commonPool();
            // 提交可分解的CalTask任务
            Future<Integer> future = pool.submit(new CalculateTask(arr, 0, arr.length));
    
            consume = new Date().getTime() - now;
            System.out.println("多线程任务分解就算加和，任务结束;耗时：" + consume);
    
            System.out.println("多线程任务分解就算加和：" + future.get());
    
            // 关闭线程池
            pool.shutdown();
    
            System.out.println("--------------------------------------");
        }
    }

# 4、多线程任务分解测试结果 #

--------------------------------------
    单线程循环遍历累计加和，任务开始...
    单线程循环遍历累计加和，任务结束;耗时：8
    单线程循环遍历累计加和：95001024
    --------------------------------------
    多线程任务分解就算加和，任务开始...
    多线程任务分解就算加和，任务结束;耗时：4
    多线程任务分解就算加和：95001024
    --------------------------------------

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5pil6aOO5YyW5L2c56eL6Zuo_size_10_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/1dc92085feb240c9867ef52c0b150d42.png