Java并行计算Fork/Join框架使用

谁借莪１个温暖的怀抱￠ 2022-08-21 08:51 207阅读 0赞

## 背景介绍 ##

假如目前有个需求，计算1000个数字之和，此需求是不是很简单，一次循环，即可完成计算；但如果是计算100W甚至更多的呢？当然，此时的循环依然可以达到目的，但效率就不敢恭维；同时，如果此时有个需求，需要统计100个文件中某个单词出现的次数呢？最直接的办法也是依次循环这100个文件，最终统计到结果，更好一步，你应该想到了线程池处理，起10个线程，每个线程读10个文件统计，这样效率就提升10倍左右。使用线程池可以很好的解决问题，但是对最终结果合并可能比较麻烦。幸好，JDK为我们提供了很好的并行解决方案fork/join框架。

## Fork/Join介绍 ##

Fork/Join主要功能即将一个任务拆分为多个子任务执行，最终再将子任务的执行结果合并（了解Hadoop的童鞋此时应该想起了Hadoop的MapReduce）。  
如果需要学习Fork/Join，首先需要了解如下四个类：  
\- ForkJoinPoll：程序中只需要创建一个这样的实例来运行所有fork-join任务。  
\- RecursiveTask：在线程池中运行一个本类的子类，它可以返回结果。  
\- RecursiveAction：和RecursiveTask类似但是不返回结果。  
\- ForkJoinTask：是RecursiveTask和RecursiveAction的父类，join方法就是定义在本类中。一般不需要直接使用此类，但是如果想要了解更多方法，你可以在本类中找到很多有用的javadoc文档。

## Fork/Join图例介绍 ##

![Fork/Join图例介绍][Fork_Join]

**分治法原理**  
![分治法原理][20160425165142932]

## 代码示例 ##

我们以计算10000个数字和为例说明Fork/Join使用方法：

public class Sum extends RecursiveTask<Long>
    {
    
        /** * {变量说明} */
        private static final long serialVersionUID = 1L;
        private static final Long THRESHOLD = 100L;
        private Long start;
        private Long end;
    
        public Sum(Long start, Long end)
        {
            this.start = start;
            this.end = end;
        }
    
        @Override
        protected Long compute()
        {
            Long sum = 0L;
            if ((end - start) < THRESHOLD)
            {
                for (Long i = start; i <= end; i++)
                {
                    sum += i;
                }
            } else
            {
    // System.out.println("fork");
                Long middle = (start + end) / 2;
                Sum left = new Sum(start, middle);
                Sum right = new Sum(middle + 1, end);
                left.fork();
                right.fork();
    
                sum = left.join() + right.join();
            }
            return sum;
        }
    
    }

> 代码比较简单，Sum继承了RecursiveTask，RecursiveTask如上面所说，会返回结果，此处我们是计算和，所以需要将计算结果返回；  
> 代码主要逻辑：如果待计算集合小于100时，则循环求和即可，反之，则将任务拆分为两个子任务(fork)，然后再将两个任务结果合并（join），最终返回两个任务的合并结果。

测试：

public class CommonTest
    {
        public static void main(String[] args) throws Exception
        {
            ForkJoinPool pool = new ForkJoinPool();
            Future<Long> result = pool.submit(new Sum(0L, 10000L));
    // pool.shutdown(); 
            System.out.println(result.get());
    
        }
    }

使用起来是不是很简单，这就是JDK的Fork/Join模式。

**注意：**  
ForkJoinPool为我们提供了shutdown()和shutdownNow()方法。  
其中执行shutdown()方法之后，ForkJoinPool 不再接受新的任务，但是已经提交的任务可以继续执行，此时如果再提交任务，则抛出异常：java.util.concurrent.RejectedExecutionException。  
而执行shotdownNow()方法后，则ForkJoinPool立即停止执行，同时抛出异常：java.util.concurrent.CancellationException。

## 后记 ##

从上面我们可知，将一个任务拆分为多个子任务去执行，可以很好的提升执行效率，但是并不是拆分的子任务越多执行效率越快，因为拆分子任务也是基于多线程思想执行，而系统中创建线程本来就是一种比较消耗时间和系统资源的。那么，在需要拆分任务的时候，应该拆分多少个任务才合理呢？参考如下算法：

> 常见方法——计算密集型，设为CPU个数+1；IO密集型，设为2\*CPU个数+1  
> 精确计算——（ IO等待时间/CPU计算时间 + 1 ）\* CPU个数

参考：  
[http://www.ibm.com/developerworks/cn/java/j-lo-forkjoin/index.html][http_www.ibm.com_developerworks_cn_java_j-lo-forkjoin_index.html]

[Fork_Join]: /images/20220724/23cc1e8d919447f2831474d3be446b5a.png
[20160425165142932]: /images/20220724/0d9af1e4c2e542c1ba391f74b57cf93f.png
[http_www.ibm.com_developerworks_cn_java_j-lo-forkjoin_index.html]: http://www.ibm.com/developerworks/cn/java/j-lo-forkjoin/index.html