大家好,又见面了,我是你们的朋友全栈君。
Fork/Join框架是Java 7提供的一个用于并行执行任务的框架,是一个把大任务分割成若干个小任务,最终汇总每个小任务结果后得到大任务结果的框架。Fork/Join框架要完成两件事情:
1.任务分割:首先Fork/Join框架需要把大的任务分割成足够小的子任务,如果子任务比较大的话还要对子任务进行继续分割
2.执行任务并合并结果:分割的子任务分别放到双端队列里,然后几个启动线程分别从双端队列里获取任务执行。子任务执行完的结果都放在另外一个队列里,启动一个线程从队列里取数据,然后合并这些数据。
在Java的Fork/Join框架中,使用两个类完成上述操作
1.ForkJoinTask:我们要使用Fork/Join框架,首先需要创建一个ForkJoin任务。该类提供了在任务中执行fork和join的机制。通常情况下我们不需要直接集成ForkJoinTask类,只需要继承它的子类,Fork/Join框架提供了两个子类:
a.RecursiveAction:用于没有返回结果的任务
b.RecursiveTask:用于有返回结果的任务
2.ForkJoinPool:ForkJoinTask需要通过ForkJoinPool来执行
任务分割出的子任务会添加到当前工作线程所维护的双端队列中,进入队列的头部。当一个工作线程的队列里暂时没有任务时,它会随机从其他工作线程的队列的尾部获取一个任务(工作窃取算法)。
参照链接地址:https://www.cnblogs.com/senlinyang/p/7885964.html
package com.company.demo1;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
/**
* fork join学习
*
* 以下计算 1 + 2 + 3 + 4 + ... + n 的结果
*
* <p>
* RecursiveTask 和 RecursiveAction 的区别就是前者有返回值,后者无返回值
* <p>
* 通过这个例子,进一步了解ForkJoinTask,ForkJoinTask与一般任务的主要区别在于它需要实现compute方法,
* 在这个方法里,首先需要判断任务是否足够小,如果足够小就直接执行任务。
* 如果不足够小,就必须分割成两个子任务,每个子任务在调用fork方法时,又会进入compute方法,
* 看看当前子任务是否需要继续分割成子任务,如果不需要继续分割,则执行当前子任务并返回结果。
* <p>
* 使用join方法会等待子任务执行完并得到其结果。
*/
public class ForkJoinTest extends RecursiveTask<Integer> {
private static final int THREAD_HOLD = 2;
private int start;
private int end;
private ForkJoinTest(int start, int end) {
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
int sum = 0;
//如果任务足够小就计算
boolean canCompute = (end - start) <= THREAD_HOLD;
if (canCompute) {
for (int i = start; i <= end; i++) {
sum += i;
}
//模拟耗时计算
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("任务足够小了,直接计算,计算结果:" + sum);
} else {
int middle = (start + end) / 2;
System.out.println("任务还不够小,继续分割,middle:" + middle);
ForkJoinTest left = new ForkJoinTest(start, middle);
ForkJoinTest right = new ForkJoinTest(middle + 1, end);
//执行子任务
left.fork();
right.fork();
//获取子任务结果
long startTime = System.currentTimeMillis();
//join() 方法会阻塞到结果算出来
int lResult = left.join();
int rResult = right.join();
System.out.println("当前线程:" + Thread.currentThread().getName() +"耗时:" + (System.currentTimeMillis() - startTime) + "ms");
sum = lResult + rResult;
System.out.println("计算结果 sum:" + sum);
}
return sum;
}
public static void main(String[] args) {
ForkJoinPool pool = new ForkJoinPool();
ForkJoinTest task = new ForkJoinTest(1, 100);
Future<Integer> result = pool.submit(task);
try {
long start = System.currentTimeMillis();
Integer sum = result.get();
System.out.println("耗时:" + (System.currentTimeMillis() - start) + "ms,计算结果:" + sum);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
}
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
发布者:全栈程序员栈长,转载请注明出处:https://javaforall.cn/191230.html原文链接:https://javaforall.cn