JUC之ForkJoin框架

ForkJoin

ForkJoin是由JDK1.7后提供多线并发处理框架, ForkJoin的框架的基本思想是分而治之。使用ForkJoin将相同的计算任务通过多线程的进行执行, 从而能提高数据的计算速度。在google的中的大数据处理框架mapreduce就通过类似ForkJoin的思想, 通过多线程提高大数据的处理。

使用ForkJoin框架, 需要创建一个ForkJoin的任务。因为ForkJoin框架为我们提供了RecursiveAction和RecursiveTask。我们只需要继承ForkJoin为我们提供的抽象类的其中一个并且实现compute方法。

RecursiveTask在进行exec之后会使用一个result的变量进行接受返回的结果。
而RecursiveAction在exec后是不会保存返回结果。

分而治之就是将一个复杂的计算, 按照设定的阈值进行分解成多个计算, 然后将各个计算结果进行汇总。相应的ForkJoin将复杂的计算当做一个任务, 而分解的多个计算则是当做一个子任务。

ForkJoinPool

Task要通过ForkJoinPool来执行, 的子任务也会添加到当前工作线程的双端队列中, 进入队列的头部。当一个工作线程中没有任务时, 会从其他工作线程的队列尾部获取一个任务(工作窃取)。

public static void main(String[] args) throws ExecutionException, InterruptedException {//创建MyTask对象MyTask myTask = new MyTask(0, 100);//创建分支合并池对象ForkJoinPool forkJoinPool = new ForkJoinPool();//获取合并之后非结果ForkJoinTask<Integer> submit = forkJoinPool.submit(myTask);Integer result = submit.get();System.out.println(result);//关闭线程池对象forkJoinPool.shutdown();
}

工作窃取(work-stealing)

任务进行分解成多个子任务的时候，每个子任务的处理时间都不一样。

例如分别有子任务A和B。如果子任务A的1ms的时候已经执行，子任务B还在执行。那么如果子任务A的线程等待子任务B完毕后在进行汇总，那么子任务A线程就会在浪费执行时间，最终的执行时间就以最耗时的子任务为准。

而如果子任务A执行完毕后，处理子任务B的任务，并且执行完毕后将任务归还给子任务B。这样就可以提高执行效率，这就是工作窃取。

案例1: 1-100的和

使用二分法将100不断拆分, 直到每个线程的任务都相对轻松为止, 直接看代码:

//写这段代码的人肯定没听说过高斯的故事
public class ForkJoinDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {//创建MyTask对象MyTask myTask = new MyTask(0, 100);//创建分支合并池对象ForkJoinPool forkJoinPool = new ForkJoinPool();//获取合并之后非结果ForkJoinTask<Integer> submit = forkJoinPool.submit(myTask);Integer result = submit.get();System.out.println(result);//关闭线程池对象forkJoinPool.shutdown();}
}class MyTask extends RecursiveTask<Integer> {//拆分值不能超过10, 防止线程过度创建private static final int VALUE = 10;private int begin;private int end;private int result = 0;public MyTask(int begin, int end) {this.begin = begin;this.end = end;}@Overrideprotected Integer compute() {if ((end - begin) <= VALUE) {for (int i = begin; i <= end; i++) {result += i;}} else {int middle = (begin + end) / 2;MyTask myTask1 = new MyTask(begin, middle);MyTask myTask2 = new MyTask(middle + 1, end);myTask1.fork();myTask2.fork();result = result + myTask1.join() + myTask2.join();}return result;}
}

案例2: 八皇后问题

八皇后问题是一个古老而又著名的问题, 是学习回溯算法的一个经典案例。

在8×8格的国际象棋上摆放八个皇后, 使其不能互相攻击, 即任意两个皇后都不能处于同一行、同一列或同一斜线上, 问一共有多少种摆法。

我们这里使用Fork Join框架并行解决问题, 加快处理速度。

上代码:

//0为棋盘, 8为皇后, 1为皇后攻击位
public class ForkJoinDemo2 {public static void main(String[] args) throws InterruptedException {int[][] chessboard = new int[8][8];ForkJoinPool forkJoinPool = new ForkJoinPool();forkJoinPool.submit(new EightQueen(0, chessboard));//等待任务完成forkJoinPool.awaitTermination(2, TimeUnit.SECONDS);forkJoinPool.shutdown();}
}class EightQueen extends RecursiveAction {private static int count = 0;int row;int[][] chessboard;public EightQueen(int row, int[][] chessboard){this.row = row;this.chessboard = chessboard;}//打印棋盘, 这里一定要加把锁, 不然打印会乱套private synchronized static void printChessboard(int[][] chessboard) {System.out.println("----第" + (++count) + "种解法----");for (int row = 0; row < chessboard.length; row++) {for (int col = 0; col < chessboard[row].length; col++) {if (chessboard[row][col] == 1) chessboard[row][col] = 0;//将被判定为攻击位的地方重新初始化为*System.out.print(chessboard[row][col]);System.out.print(' ');}System.out.println();}}//把攻击位设置为1private static void putAttack(int nowRow, int nowCol, int[][] chessboard) {for (int row = 0; row < chessboard.length; row++) {chessboard[row][nowCol] = 1;for (int col = 0; col < chessboard[row].length; col++) {chessboard[nowRow][col] = 1;if (row + col == nowRow + nowCol) chessboard[row][col] = 1;if (row - col == nowRow - nowCol) chessboard[row][col] = 1;}}chessboard[nowRow][nowCol] = 8;//上面的方法会把皇后位修改成攻击位，这里需要复原成皇后位}@Overrideprotected void compute() {if (row == chessboard.length) {//如果行数等于length，说明最后一行放置完毕，打印棋盘并退出方法printChessboard(chessboard);return;}for (int col = 0; col < chessboard[row].length; col++) {if (chessboard[row][col] == 0) {//创建备份, 等待分支结束后数组复原int[][] backup = new int[chessboard.length][chessboard.length];for (int i = 0; i < chessboard.length; i++) {for (int j = 0; j < chessboard[i].length; j++) {backup[i][j] = chessboard[i][j];}}chessboard[row][col] = 8;putAttack(row, col, chessboard);//开启分支, 进入下一行EightQueen eightQueen = new EightQueen(row + 1, chessboard);eightQueen.fork();//分支结束, 数组复原chessboard = backup;}}}
}

使用提示

使用这种多线程带来的数据共享问题, 在处理结果的合并的时候如果涉及到数据共享的问题, 我们尽可能使用JDK为我们提供的并发容器。
在使用JVM的时候我们要考虑OOM的问题, 如果我们的任务处理时间非常耗时, 并且处理的数据非常大的时候会造成OOM。
ForkJoin也是通过多线程的方式进行处理任务。, 那么我们不得不考虑是否应该使用ForkJoin。因为当数据量不是特别大的时候, 我们没有必要使用ForkJoin, 多线程会涉及到上下文的切换, 所以数据量不大的时候使用串行比使用多线程快。

我的个人主页: www.ayu.link
本文连接: ┏ (゜ω゜)=☞

JUC之ForkJoin框架相关推荐

谈谈 ForkJoin 框架的设计与实现
在了解Fork-Join之前,我们得先了解什么是并行计算. 并行计算相对于串行计算,并行计算可以划分成时间并行和空间并行.时间并行即指令流水化,也就是流水线技术.比如说生产一辆小汽车,有特定的轮子车 ...
分支合并 Fork-Join 框架
一.什么是 Fork-Join Fork/Join框架是Java7提供了的一个用于并行执行任务的框架,是一个把大任务分割成若干个小任务,最终汇总每个小任务结果后得到大任务结果的框架,这种开发方法也叫分 ...
Java7任务并行执行神器：ForkJoin框架
转载自 Java7任务并行执行神器:Fork&Join框架 Fork/Join是什么? Fork/Join框架是Java7提供的并行执行任务框架,思想是将大任务分解成小任务,然后小任务又可以继 ...
java forkjoin MySQL_Java并发fork-join框架
fork-join框架允许在几个工作进程中断某个任务,然后等待结果组合它们. 它在很大程度上利用了多处理器机器的生产能力. 以下是fork-join框架中使用的核心概念和对象. Fork Fork是一 ...
Java ForkJoin 框架初探
多核时代,编程语言如果不支持多核编程就OUT了,Java为了迎头赶上,在Java 8 版本增加大量支持多核编程的类库,如Stream等,Java 7开始支持的ForkJoin框架也是为了更好的支持多核 ...
ForkJoin框架源码分析（详细）
ForkJoin简介及使用 ForkJoin框架是CompletableFuture和java8 stream使用到的框架.主要用于分片处理的场景. 可以通过自定义分片粒度来实现任务分解.并行处理数据 ...
ForkJoin框架详解一张图搞明白工作窃取(work-stealing)机制
1 ForkJoin框架 1.1 ForkJoin框架 ForkJoinPool一种ExecutorService的实现,运行ForkJoinTask任务.ForkJoinPool区别于其它Execu ...
使用forkjoin框架分页查询所有数据的例子
使用forkjoin框架分页查询所有数据的例子 import io.swagger.annotations.ApiOperation; import lombok.AllArgsConstructor ...
JUC系列（十） | ForkJoin框架并行处理任务
多线程一直Java开发中的难点,也是面试中的常客,趁着还有时间,打算巩固一下JUC方面知识,我想机会随处可见,但始终都是留给有准备的人的,希望我们都能加油!!! 沉下去,再浮上来,我想我们会变的不一样 ...