8.3分叉/结合的框架（Fork/Join Framework）

我信总是需要让我们的代码执行更快。过去，我们需要依赖于微处理器的速度提升和通过提供多处理器。然而，大概在2003年，微处理器的速度停止提升，因为自然因素的限制。为了弥补这个不足，我们需要人工的处理器去添加多处理核，这样就可以在高度并发性中提升速度。
Java通过低级的线程提供并发的应用，和高级的并发工具应用如线程池。问题是并发不能最大的使用变量处理器或核的资源。例如，假定你创建一个排序的算法，将一个要排序的数组分隔为两部分，设计两个线程分别去处理这两部分，然后再将两线程处理的结果合并。
让我们假定每一个线程的返回是在一个不同的处理器。因为不同元素的计算是在每一个分隔的数组中重新排序的。它可能出现这样的问题，当一个线程处理完成之前，可能另一个线程已经处理好了，然后在等待另一个线程执行好后再合并处理结果。在这个例子中，一个处理的资源是浪费的。
这个问题（而且代码是冗长的很难被阅读）可以通过递归的方法来解决，将一个任务分为多个子任务来执行，然后合并结果。这些子任务会在平行中运行和在大概相同的时间处理完（如果没有特殊情况），这样返回的结果就会被组合并且通过之前的子任务来返回最终的结果。这样的解决方式，在线程等待时，很难会有处理的时间被浪费，而且递归的代码简洁和容易读懂。Java提供了分叉/结合框架（Fork/Join Framework）应用这些方案中。
Fork/Join包含一个特殊的执行服务和线程池。这个执行器服务让一个任务有效的在框架中，并且让一个任务分隔成多个小的任务，这些小的任务可以通过线程池执行在不同的线程中。一个任务会在等待，直到所有子任务都完成好并返回。
Fork/Join使工作悄悄地减少线程的竞争和开销。每一个来自工作线程池的工作线程，有它自己的双向的对列和放入一个新的任务到这个对列中。它将会读取队列头的任务。如果队列是空的，那么它就会试图去其它队列的尾部去拿任务。去其它线程拿任务是罕见的，因为工作线程是按照先进后出（Last-in-First-out）的顺序的，和任务的大小获取更小的问题，还会将问题分隔为子问题的。你开始将一个任务作为核心的工作处理，它就会将任务分隔为多个小的任务。最终所有的工作都会使用更小的同步去完成任务。

Fork/Join包含着java.util.concurrent包的ForkJoinPool,ForkJoinTask,ForkJoinWorkerThread,RecursiveAction,RecusiveTask和CountedCompleter的类。
ForJoinPool是一个java.util.concurrent.ExcutorService的实施，用于ForkJoinTask。一个ForkJoinPool从non-ForkJoinTask的客户端中提供一个关键的提交，也就提供管理监测顺的操作。
ForkJoinTask是一个基础类的接口，并且返回一个ForkJoinPool的上下文。一个ForkJoinTask的实例是一个类似于实例的线程，比起普通的线程它会更加轻量。大量的任务和子任务可能在一个ForkJoinPool中持有一个小数量的计算线程，每一个执行的代价都是有限的。
ForkJoinWorkerThread描述一个通过ForkJoinPool实例操作的线程，这个会执行ForkJoinTasks.
RecursiveAction描述一个递归的结果，而不是ForkJoinTask
RecursiveTask描述一个关于ForkJoinTask的递归结果。
CountedCompleter描述一个ForkJoinTask带有竞争的动作的应用，这个状态是在触发器没有保持在挂起的动作上的。

Java的文档提供了关于RecursiveAction和RecursiveTask为基础的例子，例如在Listing 8-5的Matrix的类代表着一个矩阵包含了特殊数据的行和列。

package com.owen.thread.chapter8;public class Matrix
{private final int[][] matrix;public Matrix(int nrows, int ncols){matrix = new int[nrows][ncols];}public int getCols(){return matrix[0].length;}public int getRows(){return matrix.length;}public int getValue(int row, int col){return matrix[row][col];}public void setValue(int row, int col, int value){matrix[row][col] = value;}}

Listing8-6例子是一个单一的线程使用两Matrix的实例。

package com.owen.thread.chapter8;public class MatMult
{public static void main(String[] args){Matrix a = new Matrix(1, 3);a.setValue(0, 0, 1); // | 1 2 3 |a.setValue(0, 1, 2);a.setValue(0, 2, 3);      dump(a);Matrix b = new Matrix(3, 2);b.setValue(0, 0, 4); // | 4 7 |b.setValue(1, 0, 5); // | 5 8 |b.setValue(2, 0, 6); // | 6 9 |b.setValue(0, 1, 7);b.setValue(1, 1, 8);b.setValue(2, 1, 9);dump(b);dump(multiply(a, b));}public static void dump(Matrix m){for (int i = 0; i < m.getRows(); i++){for (int j = 0; j < m.getCols(); j++)System.out.printf("%d ", m.getValue(i, j));System.out.println();}System.out.println();}public static Matrix multiply(Matrix a, Matrix b){if (a.getCols() != b.getRows())throw new IllegalArgumentException("rows/columns mismatch");Matrix result = new Matrix(a.getRows(), b.getCols());for (int i = 0; i < a.getRows(); i++)for (int j = 0; j < b.getCols(); j++)for (int k = 0; k < a.getCols(); k++)result.setValue(i,j,result.getValue(i, j) + a.getValue(i, k)* b.getValue(k, j));return result;}}

Listing8-6的MatMult类声明了一个multiply()的方法，这个方法演示了多个矩阵的应用。之后验证在第一个Matrix(a)的列的数是否与Matrix(b)的行数相等。这个实质的算法是方法multiply()创建result Matrix和在多应用中循环放入队列。

这些循环本质遵守：对于a中的每一行，用b中相应的行值乘以该行的每一列值，然后将乘法的结果添加到存储区，并通过在a中的行索引（i）和b中列索引（j）指定的位置存储总的总结果。

执行结果：

 1 2 3 4 7
5 8
6 9 32 50

科学计算分类这个算法是O(n*n*n)。这个让我们难以想象这个分类算法的性能是如此的差。

这个应用可以被改善（在多处理和多核的平台中），将每个列按乘法任务分配给单独的线程实体。Listing8-7展示了你应该如何在上下文的Fork/Join的框架中这个方案。

package com.owen.thread.chapter8;import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;public class MatMultRecursive extends RecursiveAction
{private final Matrix a, b, c;private final int row;public MatMultRecursive(Matrix a, Matrix b, Matrix c){this(a, b, c, -1);}public MatMultRecursive(Matrix a, Matrix b, Matrix c, int row){if (a.getCols() != b.getRows())throw new IllegalArgumentException("rows/columns mismatch");this.a = a;this.b = b;this.c = c;this.row = row;}@Overridepublic void compute(){if (row == -1){List<MatMultRecursive> tasks = new ArrayList<>();for (int row = 0; row < a.getRows(); row++)tasks.add(new MatMultRecursive(a, b, c, row));invokeAll(tasks);} elsemultiplyRowByColumn(a, b, c, row);}public static void multiplyRowByColumn(Matrix a, Matrix b, Matrix c, int row){for (int j = 0; j < b.getCols(); j++)for (int k = 0; k < a.getCols(); k++)c.setValue(row,j,c.getValue(row, j) + a.getValue(row, k)* b.getValue(k, j));}public static void dump(Matrix m){for (int i = 0; i < m.getRows(); i++){for (int j = 0; j < m.getCols(); j++)System.out.print(m.getValue(i, j) + " ");System.out.println();}System.out.println();}public static void main(String[] args){Matrix a = new Matrix(2, 3);a.setValue(0, 0, 1); // | 1 2 3 |a.setValue(0, 1, 2); // | 4 5 6 |a.setValue(0, 2, 3);a.setValue(1, 0, 4);a.setValue(1, 1, 5);a.setValue(1, 2, 6);dump(a);Matrix b = new Matrix(3, 2);b.setValue(0, 0, 7); // | 7 1 |b.setValue(1, 0, 8); // | 8 2 |b.setValue(2, 0, 9); // | 9 3 |b.setValue(0, 1, 1);b.setValue(1, 1, 2);b.setValue(2, 1, 3);dump(b);Matrix c = new Matrix(2, 2);ForkJoinPool pool = new ForkJoinPool();pool.invoke(new MatMultRecursive(a, b, c));dump(c);}}

Listing8-7的例子出现了一个MatMult的类，并且继承了RecursiveAction。为了去完成这个有意思的工作，RecursiveAction的void compute()方法就会重写。

Note 尽管compute()方法是正常作用在子分隔的任务中并使用递归，我选择不同的方法去处理乘法任务。

之后创建Matrixes的a和b，在Listing8-7main()的方法中创建Matrix c和实例ForkJoinPool。然后它实例化MatMult，通过三个Matrix实例去构造MatMult(Matrix a, Matrix b,Matrix c),和请求ForkJoinPool的 Tinvoke(ForkJoinTask<T> task)方法去开始运行这个初始的任务。这个方法并不会返回直到它的初始任务和所有的子任务执行好。

MatMult(Matrix a, Matrix b, Matrix c)的构造器去执行MatMult(Matrixa, Matrix b, Matrix c)的构造器，将-1作为row的特殊值。这个值将会通过compute()的方法运用，这个执行就是前面invoke()方法请求的，在初始任务和子任务之间区分开来。

当compute()初始化被请求时（row等于-1）,它创建一个MatMult任务的列表，和通过这个列表去调用RecursiveAction的Collection<T>的invokeAll(Collection<T> tasks)的方法（继承于ForkJoinTask）。这个方法分叉List的集合中的任务，它将会开始执行。它将会处于等待状态，直到invokeAll()方法（它将会集合所有这些任务）返回。这就是会发生在boolean isDone()的方法（也是继承于ForkJoinTask）在每一个子任务中返回true。

注意tasks.add(newMatMult(a, b, c));方法的调用。这个请求注册一个特殊的row值给MatMult的实例。当invokeAll()被请求，每一个任务的compute()方法被请求，和发现不同的值（其它的值大于-1）注册到row.它会执行multiRowByColum(a, b, c, row);在特殊的row情况下执行。

8.3分叉/结合的框架（Fork/Join Framework）相关推荐

join left 大数据_Java并发编程笔记-JDK内置并行执行框架Fork/Join
Fork/Join由来-分而治之思想分而治之:对于一个比较复杂的任务,如果可以很自然地将其分解为多个子任务,这些子任务互相独立且与原问题性质相同,递归地处理这些子任务,然后将各个子任务的结果合并得到 ...
Java并行任务框架Fork/Join
Fork/Join是什么? Fork意思是分叉,Join为合并.Fork/Join是一个将任务分割并行运行,然后将最终结果合并成为大任务的结果的框架,父任务可以分割成若干个子任务,子任务可以继续分割, ...
JUC组件扩展（二）-JAVA并行框架Fork/Join（四）：监控Fork/Join池
Fork/Join 框架是为了解决可以使用 divide 和 conquer 技术,使用 fork() 和 join() 操作把任务分成小块的问题而设计的.主要实现这个行为的是 ForkJoinPoo ...
fork join框架_Fork / Join框架vs.并行流vs.ExecutorService：最终的Fork / Join基准
fork join框架 Fork / Join框架在不同配置下如何工作? 就像即将到来的<星球大战>(Star Wars)一样,围绕Java 8并行性的批评也充满了兴奋. 并行流的语法糖带 ...
Fork / Join框架vs并行流vs.ExecutorService：最终的Fork / Join基准
Fork / Join框架在不同配置下如何工作? 就像即将上映的<星球大战>一样,围绕Java 8并行性的批评也充满了兴奋. 并行流的语法糖带来了一些炒作,就像我们在预告片中看到的新型光剑 ...
fork join框架使用_Java：使用Fork / Join框架的Mergesort
fork join框架使用此项的目的是显示一个Fork / Join RecursiveAction的简单示例,而不是过多地探讨合并排序的可能优化方法,或者比使用Exkutor / Join Poo ...
Java 7：满足Fork / Join框架
JSR-166(y)是Java 7中包含的此新功能的正式名称.如果您发现名称中有一个" y",这是因为自Java 5起就添加了JSR-166(并发实用程序) ,但它不会就此停止,因 ...
Java：使用Fork / Join框架的Mergesort
此项的目的是显示一个Fork / Join RecursiveAction的简单示例,而不是过多地研究合并合并的可能优化方法,或者比使用Exkutor / Join Pool优于现有的基于Java 6 ...
java fork_浅谈Java的Fork/Join并发框架
前几天有写到整合并发结果的文章,于是联想到了Fork/Join.因为在我看来整合并发结果其实就是Fork/Join中的Join步骤.所以今天我就把自己对Fork/Join一些浅显的理解记录下来. 1. ...
浅谈Java的Fork/Join并发框架
1. Fork/Join是什么 Oracle的官方给出的定义是:Fork/Join框架是一个实现了ExecutorService接口的多线程处理器.它可以把一个大的任务划分为若干个小的任务并发执行,充 ...

8.3分叉/结合的框架（Fork/Join Framework）

8.3分叉/结合的框架（Fork/Join Framework）相关推荐

最新文章

热门文章