java 多线程线程池：多核CPU利用ExecutorService newWorkStealingPool； ForkJoinPool线程池执行可拆分的任务RecursiveAction；RecursiveTask

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

1，给定并行级别：

1， ExecutorService newWorkStealingPool(int parallelism): 创建持有足够的线程的线程池来支持给定的并行级别，该方法还会使用多个队列来减少竞争
2， ExecutorService newWorkStealingPool() : 该方法是前面方法的简化版本如果前机器有4个CPU，则目标并行级别被设置为4

这两个方法是Java8新增的，这两个方法可充分利用多 CPU 并行的能力这两个方法生成的 work stealing 池，都相于后台线程池，如果所有的前台线程都死亡了workstealing 池中的线程会自动死亡。

用法：

通过Executors.newWorkStealingPool()静态方法获取ExecutorService类对象
使用ExecutorService对象.submit(Runnable runnable)提交runnable 或Callable 接口类实现的任务对象

示例代码：

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

/**

 * @ClassName ExecutorsWorkStealingPoolTest

 * @projectName: object1

 * @author: Zhangmingda

 * @description： XXX

 * date: 2021/4/27.

 */

public class ExecutorsWorkStealingPoolTest {

    public static void main(String[] args) throws InterruptedException {

        Runnable r = () -> {

            String tName = Thread.currentThread().getName();

            try {

                System.out.println(tName + "开始运行");

                Thread.sleep(1000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println(tName + "运行结束");

        };

        /**

         * 创建支持多核CPU并行的线程池

         */

        ExecutorService executorService = Executors.newWorkStealingPool(); //译文：Stealing 窃取

        for (int i=0; i<10; i++){

            executorService.submit(r);

        }

        System.out.println(Runtime.getRuntime().availableProcessors()); //CPU核心数

        Thread.sleep(3000);

    }

}

2、Java8 增强的 ForkJoinPool 用于拆分大的计算任务，拆分为多个小的计算任务
为了充分利用CPU、多核CPU的性能优势，计算机软件系统应该可以充分"挖掘"每个CPU的计算能力，绝不能让某个CPU处于"空闲"状态，为了充分利用多CPU、多核CPU的优势，可以考虑把一个任务拆分成多个"小任务"，把多个"小任务"放到多个处理器核上并行执行;当多个"小任务"执行完成之后，再将这些执行结果合并起来即可。 ForkJoinPool非常适合做密集计算型的任务。

用法：

定义可以拆分的任务类，继承RecursiveAction类或RecursiveTask类(可以有返回值) 实现其抽象方法compute；方法中自调用创建子 任务对象.fork()方法提交拆分任务
用自定义任务类，实例化可拆分任务对象；子任务类的join()方法获取返回值
使用ForkJoinPool实例化对象，submit可拆分任务对象；返回 ForkJoinTask对象
ForkJoinPool对象awaitTermination执行任务(无返回值)；。
使用 ForkJoinTask对象 对象.get()方法获取返回值

无返回值示例代码（ RecursiveAction类任务）：

要求：给出一个int数据范围，范围内整数个数>100 则拆分不同的线程来打印。每个线程只打印不超过100个整数。

import java.util.concurrent.ForkJoinPool;

import java.util.concurrent.RecursiveAction;

import java.util.concurrent.TimeUnit;

/**

 * @ClassName MyRecursiveAction

 * @projectName: object1

 * @author: Zhangmingda

 * @description： XXX

 * date: 2021/4/27.

 */

public class MyRecursiveAction {

    /**

     * 定义一个支持拆分计算的任务

     */

    private static class PrintTaskRecursiveAction extends RecursiveAction{

        private int start;

        private int end;

        private final int MAXNUM = 100;

        /**

         * 构造实例传入任务需要的参数

         */

        public PrintTaskRecursiveAction(int start, int end) {

            this.start = start;

            this.end = end;

        }

        /**

         * 具体执行计算的任务的抽象方法重写

         */

        @Override

        protected void compute() {

            String tName = Thread.currentThread().getName();

            if ((end - start) < MAXNUM){

                System.out.println(tName + "    start:" + start);

                System.out.println(tName + "    end:" + end);

            }else {

                int middle = (start + end) /2;

                /**

                 * 大任务拆分为两个小任务，

                 */

                PrintTaskRecursiveAction subTask1 = new PrintTaskRecursiveAction(start,middle);

                PrintTaskRecursiveAction subTask2 = new PrintTaskRecursiveAction(middle,end);

                //分别执行两个小任务

                subTask1.fork();subTask2.fork();

            }

        }

    }

    /**

     执行计算任务

     */

    public static void main(String[] args) throws InterruptedException {

        ForkJoinPool forkJoinPool = new ForkJoinPool();

        PrintTaskRecursiveAction printTask = new PrintTaskRecursiveAction(1,300);

        //线程池提交任务

        forkJoinPool.submit(printTask);

        forkJoinPool.awaitTermination(1, TimeUnit.SECONDS);

        //关闭提交接口

        forkJoinPool.shutdown();

    }

}

有返回值示例代码（ RecursiveTask类任务 ）：

要求：计算1~100的和，每个线程计算不超过10个数的和。

import java.util.concurrent.*;

/**

 * @ClassName MyRecursiveAction

 * @projectName: object1

 * @author: Zhangmingda

 * @description： XXX

 * date: 2021/4/27.

 */

public class ForkJoinPoolRecursiveTasksReturnExample {

    /**

     * 定义一个支持拆分计算的任务

     */

    private static class CalcNumCountTaskRecursiveTask extends RecursiveTask<Integer> {

        private int start;

        private int end;

        private final int MAXNUM = 30;

        /**

         * 构造实例传入任务需要的参数

         */

        public CalcNumCountTaskRecursiveTask(int start, int end) {

            this.start = start;

            this.end = end;

        }

        /**

         * 具体执行计算的任务的抽象方法重写

         */

        @Override

        protected Integer compute() {

            String tName = Thread.currentThread().getName();

            Integer count = 0;

            if ((end - start) < MAXNUM){

                System.out.println("start:" + start);

                System.out.println("end:" + end);

                for (int i=start; i<end; i++){

                    count+=i;

                }

                return count;

            }else {

                int middle = (start + end) /2;

                /**

                 * 大任务拆分为两个小任务，

                 */

                CalcNumCountTaskRecursiveTask subTask1 = new CalcNumCountTaskRecursiveTask(start,middle);

                CalcNumCountTaskRecursiveTask subTask2 = new CalcNumCountTaskRecursiveTask(middle,end);

                //分别执行两个小任务

                subTask1.fork();subTask2.fork();

                return subTask1.join() + subTask2.join();

            }

        }

    }

    /**

     执行计算任务

     */

    public static void main(String[] args) throws InterruptedException, ExecutionException {

        ForkJoinPool forkJoinPool = new ForkJoinPool();

        CalcNumCountTaskRecursiveTask calcCountTask = new CalcNumCountTaskRecursiveTask(1,101);

        //线程池提交任务

        ForkJoinTask<Integer> forkJoinTask =  forkJoinPool.submit(calcCountTask);

        //获取执行结果

        System.out.println(forkJoinTask.get());;

        //关闭提交接口

        forkJoinPool.shutdown();

    }

}

正文

用法：

无返回值示例代码（ RecursiveAction类任务）：

相关阅读

jquery图片url拆分，jqueryeach

js数组拆分数组，js将数组分成n段操作

hbaseregion拆分问题，hbase 分区表

mysql表数据切割，mysql表拆分

面向对象模块拆分，面向对象分解

hbase啥时候开始拆分，hbase组成

js中树数据结构怎么拆分，js实现树结构菜单

go语言将pdf文件拆分，go解析pdf

目录[+]

用法：

无返回值示例代码（ RecursiveAction类任务 ）：

相关阅读

jquery图片url拆分，jqueryeach

js数组拆分数组，js将数组分成n段操作

hbaseregion拆分问题，hbase 分区表

mysql表数据切割，mysql表拆分

面向对象模块拆分，面向对象分解

hbase啥时候开始拆分，hbase组成

js中树数据结构怎么拆分，js实现树结构菜单

go语言将pdf文件拆分，go解析pdf

目录[+]

无返回值示例代码（ RecursiveAction类任务）：