forkjoinpool_forkjoinpool 队列_forkjoinpool源码分析

我们常常听说CPU拥有多核，但它对执行程序有什么大大的帮助呢？相对于CPU单核，多核CPU可以把一个任务比较大的程序分成几个任务比较小的程序，让这些小程序放到处理器上的多个核心同时处理，最后再把这些执行结果合并起来，这样可以大大提升运行效率，就拿个生活中的例子来描述CPU多核的用处吧：你去大学食堂打饭时，食堂如果仅仅有一个打饭窗口，我们学生是不是要排很长的队伍去等待，直到前面的人打完饭，才轮到你了，但如果食堂加了几个打饭窗口，那么我们这些可以分散到几个窗口去打饭，就不用排那么长的队伍了，这样你打到饭的时间是不是少了很多啊，这里打饭窗口就类似于多核CPU的用处。

Java就提供了一个ForkJoinPool类来支持并行处理程序。

代码例子：

**没有返回值得并行计算**
//继承RecursiveAction来实现"可分解"的任务
class PrintTask extends RecursiveAction
{
    // 每个“小任务”只最多只打印50个数
    private static final int THRESHOLD = 10;
    private int start;
    private int end;
    // 打印从start到end的任务
    public PrintTask(int start, int end)
    {
        this.start = start;
        this.end = end;
    }
    @Override
    protected void compute() 
    {
        // 当end与start之间的差小于THRESHOLD时，开始打印
        if(end - start < THRESHOLD)
        {
            for (int i = start ; i < end ; i++ )
            {
                System.out.println(Thread.currentThread().getName()
                    + "的i值：" + i);
            }
        }
        else
        {
            // 如果当end与start之间的差大于THRESHOLD时，即要打印的数超过50个
            // 将大任务分解成两个小任务。
            int middle = (start + end) /2;
            PrintTask left = new PrintTask(start, middle);
            PrintTask right = new PrintTask(middle, end);
            // 并行执行两个“小任务”
            left.fork();
            right.fork();
        }
    }
}
public class ForkJoinPoolTest
{
    public static void main(String[] args) 
        throws Exception
    {
        ForkJoinPool pool = new ForkJoinPool();
        // 提交可分解的PrintTask任务
        pool.submit(new PrintTask(0 , 50));
        //线程阻塞，等待所有任务完成
        pool.awaitTermination(2, TimeUnit.SECONDS);
        // 关闭线程池
        pool.shutdown();
    }
}

运行结果：

ForkJoinPool-1-worker-3的i值：18

ForkJoinPool-1-worker-4的i值：37

ForkJoinPool-1-worker-1的i值：43

ForkJoinPool-1-worker-1的i值：44

ForkJoinPool-1-worker-1的i值：45

ForkJoinPool-1-worker-1的i值：46

ForkJoinPool-1-worker-1的i值：47

ForkJoinPool-1-worker-1的i值：48

ForkJoinPool-1-worker-1的i值：49

ForkJoinPool-1-worker-1的i值：25

ForkJoinPool-1-worker-1的i值：26

ForkJoinPool-1-worker-1的i值：27

ForkJoinPool-1-worker-1的i值：28

ForkJoinPool-1-worker-1的i值：29

ForkJoinPool-1-worker-1的i值：30

ForkJoinPool-1-worker-1的i值：6

ForkJoinPool-1-worker-1的i值：7

ForkJoinPool-1-worker-1的i值：8

ForkJoinPool-1-worker-1的i值：9

ForkJoinPool-1-worker-2的i值：31

ForkJoinPool-1-worker-1的i值：10

ForkJoinPool-1-worker-4的i值：38

ForkJoinPool-1-worker-3的i值：19

ForkJoinPool-1-worker-4的i值：39

ForkJoinPool-1-worker-4的i值：40

ForkJoinPool-1-worker-1的i值：11

ForkJoinPool-1-worker-2的i值：32

ForkJoinPool-1-worker-1的i值：0

ForkJoinPool-1-worker-4的i值：41

ForkJoinPool-1-worker-3的i值：20

ForkJoinPool-1-worker-4的i值：42

ForkJoinPool-1-worker-1的i值：1

ForkJoinPool-1-worker-2的i值：33

ForkJoinPool-1-worker-1的i值：2

ForkJoinPool-1-worker-4的i值：12

ForkJoinPool-1-worker-3的i值：21

ForkJoinPool-1-worker-4的i值：13

ForkJoinPool-1-worker-1的i值：3

ForkJoinPool-1-worker-2的i值：34

ForkJoinPool-1-worker-1的i值：4

ForkJoinPool-1-worker-4的i值：14

ForkJoinPool-1-worker-3的i值：22

ForkJoinPool-1-worker-4的i值：15

ForkJoinPool-1-worker-1的i值：5

ForkJoinPool-1-worker-2的i值：35

ForkJoinPool-1-worker-2的i值：36

ForkJoinPool-1-worker-4的i值：16

ForkJoinPool-1-worker-3的i值：23

ForkJoinPool-1-worker-4的i值：17

ForkJoinPool-1-worker-3的i值：24

有返回值的并行计算

//继承RecursiveTask来实现"可分解"的任务
class CalTask extends RecursiveTask<Integer>
{
    // 每个“小任务”只最多只累加20个数
    private static final int THRESHOLD = 20;
    private int arr[];
    private int start;
    private int end;
    // 累加从start到end的数组元素
    public CalTask(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;
        }
        else
        {
            // 如果当end与start之间的差大于THRESHOLD时，即要打印的数超过20个
            // 将大任务分解成两个小任务。
            int middle = (start + end) /2;
            CalTask left = new CalTask(arr , start, middle);
            CalTask right = new CalTask(arr , middle, end);
            // 并行执行两个“小任务”
            left.fork();
            right.fork();
            // 把两个“小任务”累加的结果合并起来
            return left.join() + right.join();
        }
    }
}
public class Sum
{
    public static void main(String[] args)
        throws Exception
    {
        int[] arr = new int[100];
        Random rand = new Random();
        int total = 0;
        // 初始化100个数字元素
        for (int i = 0 , len = arr.length; i < len ; i++ )
        {
            int tmp = rand.nextInt(20);
            // 对数组元素赋值，并将数组元素的值添加到total总和中。
            total += (arr[i] = tmp);
        }
        System.out.println(total);

        ForkJoinPool pool = new ForkJoinPool();
        // 提交可分解的CalTask任务
        Future<Integer> future = pool.submit(new CalTask(arr , 0 , arr.length));
        System.out.println(future.get());
        // 关闭线程池
        pool.shutdown();
    }
}

运行结果：

969

969

本文来自电脑杂谈，转载请注明本文网址：
https://www.pc-fly.com/a/jisuanjixue/article-40530-1.html