《Java Fork/Join框架介绍》《Java Fork/Join框架介绍》

65 阅读 0 评论 43 点赞

我是靠谱客的博主糊涂溪流，最近开发中收集的这篇文章主要介绍《Java Fork/Join框架介绍》《Java Fork/Join框架介绍》，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

《Java Fork/Join框架介绍》

Fork/Join框架是Java7提供了的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。

我们已经很清楚Fork/Join框架的需求了，那么我们可以思考一下，如果让我们来设计一个Fork/Join框架，该如何设计？这个思考有助于你理解Fork/Join框架的设计。

第一步肯定是将大任务分割成小任务。首先我们需要有一个fork类来把大任务分割成子任务，有可能子任务还是很大，所以还需要不停的分割，直到分割出的子任务足够小。

第二步执行任务并合并结果。分割的子任务分别放在双端队列里，然后几个启动线程分别从双端队列里获取任务执行。子任务执行完的结果都统一放在一个队列里，启动一个线程从队列里拿数据，然后合并这些数据。

Fork/Join使用两个类来完成以上两件事情：

1、ForkJoinTask：我们要使用ForkJoin框架，必须首先创建一个ForkJoin任务。它提供在任务中执行fork()和join()操作的机制，通常情况下我们不需要直接继承ForkJoinTask类，而只需要继承它的子类，Fork/Join框架提供了以下两个子类：

RecursiveAction：用于没有返回结果的任务。
RecursiveTask ：用于有返回结果的任务。

2、ForkJoinPool：ForkJoinTask需要通过ForkJoinPool来执行，任务分割出的子任务会添加到当前工作线程所维护的双端队列中，进入队列的头部。当一个工作线程的队列里暂时没有任务时，它会随机从其他工作线程的队列的尾部获取一个任务。

通过使用 Fork/Join 模式，软件开发人员能够方便地利用多核平台的计算能力。尽管还没有做到对软件开发人员完全透明，Fork/Join 模式已经极大地简化了编写并发程序的琐碎工作。对于符合 Fork/Join 模式的应用，软件开发人员不再需要处理各种并行相关事务，例如同步、通信等，以难以调试而闻名的死锁和 data race 等错误也就不会出现，提升了思考问题的层次。你可以把 Fork/Join 模式看作并行版本的 Divide and Conquer 策略，仅仅关注如何划分任务和组合中间结果，将剩下的事情丢给 Fork/Join 框架。但是Fork/Join并行计算框架，并不是万能的，并不能解决所有应用程序在超多核心处理器上的并发问题。

如果一个应用能被分解成多个子任务，并且组合多个子任务的结果就能够获得最终的答案，那么这个应用就适合用 Fork/Join 模式来解决。我们作为程序开发者需要做的就是拆分任务并组合每个子任务的中间结果，而不用再考虑线程和锁的问题。

先看一个实际例子：求一个大数组中所有元素的和。如果我们只有一个处理器，则我们只能采用如下的方法进行：


for(int i=0;i<arr.length;i++){
sum+=arr[i];
}

如果我们有N个处理器，要想提高计算速度，则可以借助Fork/Join框架并行计算来完成。

代码如下：


import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
//利用ForkJoin框架来以每100个为单元计算大数组arr中所有元素的和
public class Calculate extends RecursiveTask<Integer>{
/**
*
*/
private static final long serialVersionUID = 1L;
private int[] arr;
private int start;
private int end;
private static final int THRESHOLD = 100;//门限
public Calculate(int[] arr){
this.arr = arr;
this.start = 0;
this.end = arr.length-1;
}
public Calculate(int[] arr,int start,int end){
this.arr = arr;
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
int interval = end-start;
int sum = 0;
if(interval<THRESHOLD){
for(int i=start;i<=end;i++){
sum+=arr[i];
}
}
else{
//任务过大，需要继续分割
int mid = start+(end-start)/2;
Calculate leftCalculate = new Calculate(arr,start,mid);
Calculate rightCalculate = new Calculate(arr,mid+1,end);
//执行子任务
leftCalculate.fork();
rightCalculate.fork();
//等待子任务执行结束，并得到子任务的结果
int leftResult = leftCalculate.join();
int rightResult = rightCalculate.join();
sum = leftResult + rightResult;
}
return sum;
}
public static void main(String[] args) {
int len = 1000;
int[] arr = new int[len];
Random r = new Random(47);
int sum = 0;
for(int i=0;i<len;i++){
arr[i] = r.nextInt(len);
sum+=arr[i];
}
ForkJoinPool forkJoinPool = new ForkJoinPool();
Calculate c = new Calculate(arr);
Future<Integer> future =
forkJoinPool.submit(c);
try {
//
forkJoinPool.awaitTermination(10, TimeUnit.SECONDS); 
int res = future.get();
System.out.println("是否相等:"+(res==sum));
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}