Java 并发 - Fork-Join 框架
fork-join 框架允许在多个工作人员上中断某个任务,然后等待结果将它们组合起来。 它很大程度上利用了多处理器机器的能力。 以下是 fork-join 框架中使用的核心概念和对象。
分叉
Fork 是一个任务将自身分割成更小的、独立的、可以并发执行的子任务的过程。
语法
Sum left = new Sum(array, low, mid); left.fork();
这里 Sum 是 RecursiveTask 的子类,left.fork() 将任务拆分为子任务。
Join
Join 是一个任务,一旦子任务执行完毕,就将所有子任务的结果进行Join,否则一直等待。
语法
left.join();
这里左边是 Sum 类的一个对象。
ForkJoinPool
它是一个特殊的线程池,设计用于处理 fork-and-join 任务分割。
语法
ForkJoinPool forkJoinPool = new ForkJoinPool(4);
这里是一个并行级别为 4 个 CPU 的新 ForkJoinPool。
RecursiveAction
RecursiveAction 代表一个不返回任何值的任务。
语法
class Writer extends RecursiveAction { @Override protected void compute() { } }
RecursiveTask
RecursiveTask 表示返回值的任务。
语法
class Sum extends RecursiveTask<Long> { @Override protected Long compute() { return null; } }
示例
以下 TestThread 程序显示了 Fork-Join 框架在基于线程的环境中的用法。
import java.util.concurrent.ExecutionException; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.RecursiveTask; public class TestThread { public static void main(final String[] arguments) throws InterruptedException, ExecutionException { int nThreads = Runtime.getRuntime().availableProcessors(); System.out.println(nThreads); int[] numbers = new int[1000]; for(int i = 0; i < numbers.length; i++) { numbers[i] = i; } ForkJoinPool forkJoinPool = new ForkJoinPool(nThreads); Long result = forkJoinPool.invoke(new Sum(numbers,0,numbers.length)); System.out.println(result); } static class Sum extends RecursiveTask<Long> { int low; int high; int[] array; Sum(int[] array, int low, int high) { this.array = array; this.low = low; this.high = high; } protected Long compute() { if(high - low <= 10) { long sum = 0; for(int i = low; i < high; ++i) sum += array[i]; return sum; } else { int mid = low + (high - low) / 2; Sum left = new Sum(array, low, mid); Sum right = new Sum(array, mid, high); left.fork(); long rightResult = right.compute(); long leftResult = left.join(); return leftResult + rightResult; } } } }
这将产生以下结果。
输出
32 499500