fork-join使用框架指南-Java快速入门教程
fork-join使用框架指南-Java快速入门教程
1. 概述
Java 7引入了fork/join框架。它提供了一些工具，通过尝试使用所有可用的处理器内核来帮助加快并行处理速度。它通过分而治之的方法实现这一目标。
在实践中，这意味着框架首先“fork”，递归地将任务分解为更小的独立子任务，直到它们足够简单，可以异步运行。
之后，“join”部分开始。所有子任务的结果都递归联接到单个结果中。对于返回 void 的任务，程序只需等待直到每个子任务运行。
为了提供有效的并行执行，fork/join 框架使用一个名为ForkJoinPool 的线程池。此池管理类型为 ForkJoinWorkerThread 的工作线程。
2.ForkJoinPool
ForkJoinPool是框架的核心。它是ExecutorService的实现，用于管理工作线程并为我们提供工具来获取有关线程池状态和性能的信息。
工作线程一次只能执行一个任务，但ForkJoinPool不会为每个子任务创建单独的线程。相反，池中的每个线程都有自己的双端队列（或deque，发音为“deck”），用于存储任务。
此体系结构对于借助工作窃取算法平衡线程的工作负载至关重要。
2.1. 工作窃取算法
简单地说，自由线程试图从繁忙线程的 deques 中“窃取”工作。
默认情况下，工作线程从其自己的 deque 的头部获取任务。当它为空时，线程从另一个繁忙线程的尾部或全局条目队列中获取任务，因为这是最大工作块可能位于的位置。
此方法最大程度地降低了线程争用任务的可能性。它还减少了线程必须去寻找工作的次数，因为它首先处理最大的可用工作块。
2.2.ForkJoinPool实例化
在Java 8中，访问ForkJoinPool实例的最方便方法是使用其静态方法commonPool(https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/ForkJoinPool.html#commonPool(%29)（）。这将提供对公共池的引用，公共池是每个ForkJoinTask 的默认线程池。
根据Oracle 的文档，使用预定义的公共池可以减少资源消耗，因为这不鼓励为每个任务创建单独的线程池。
ForkJoinPool commonPool = ForkJoinPool.commonPool();Copy我们可以在 Java 7 中通过创建一个ForkJoinPool并将其分配给实用程序类的公共静态字段来实现相同的行为：
public static ForkJoinPool forkJoinPool = new ForkJoinPool(2);Copy现在我们可以轻松访问它：
ForkJoinPool forkJoinPool = PoolUtil.forkJoinPool;Copy使用ForkJoinPool 的构造函数，我们可以创建一个具有特定并行度、线程工厂和异常处理程序级别的自定义线程池。此处池的并行级别为 2。这意味着池将使用两个处理器核心。
3.ForkJoinTask<V>
ForkJoinTask 是在 ForkJoinPool中执行的任务的基类型。实际上，应该扩展其两个子类之一：RecursiveAction用于void任务，RecursiveTask<V>用于返回值的任务。它们都有一个抽象方法compute（），其中定义了任务的逻辑。
3.1.RecursiveAction
在下面的示例中，我们使用一个名为workload的字符串来表示要处理的工作单元。出于演示目的，该任务是一个荒谬的任务：它只是将其输入大写并记录下来。
为了演示框架的fork行为，如果 workload.length（） 大于指定的阈值，该示例将使用createSubtask（） 方法拆分任务。
字符串以递归方式划分为子字符串，从而创建基于这些子字符串的自定义递归任务实例。
因此，该方法返回 List<CustomRecursiveAction>。
该列表使用invokeAll（） 方法提交到ForkJoinPool：
public class CustomRecursiveAction extends RecursiveAction {
private String workload = "";
private static final int THRESHOLD = 4;
private static Logger logger =
Logger.getAnonymousLogger();
public CustomRecursiveAction(String workload) {
this.workload = workload;
}
@Override
protected void compute() {
if (workload.length() > THRESHOLD) {
ForkJoinTask.invokeAll(createSubtasks());
} else {
processing(workload);
}
}
private List<CustomRecursiveAction> createSubtasks() {
List<CustomRecursiveAction> subtasks = new ArrayList<>();
String partOne = workload.substring(0, workload.length() / 2);
String partTwo = workload.substring(workload.length() / 2, workload.length());
subtasks.add(new CustomRecursiveAction(partOne));
subtasks.add(new CustomRecursiveAction(partTwo));
return subtasks;
}
private void processing(String work) {
String result = work.toUpperCase();
logger.info("This result - (" + result + ") - was processed by "
+ Thread.currentThread().getName());
}
}Copy我们可以使用此模式来开发我们自己的递归操作类。为此，我们创建一个表示总工作量的对象，选择合适的阈值，定义一个方法来划分工作并定义一个方法来做工作。
3.2.RecursiveTask<V>
对于返回值的任务，此处的逻辑类似。
不同之处在于，每个子任务的结果都合并在一个结果中：
public class CustomRecursiveTask extends RecursiveTask<Integer> {
private int[] arr;
private static final int THRESHOLD = 20;
public CustomRecursiveTask(int[] arr) {
this.arr = arr;
}
@Override
protected Integer compute() {
if (arr.length > THRESHOLD) {
return ForkJoinTask.invokeAll(createSubtasks())
.stream()
.mapToInt(ForkJoinTask::join)
.sum();
} else {
return processing(arr);
}
}
private Collection<CustomRecursiveTask> createSubtasks() {
List<CustomRecursiveTask> dividedTasks = new ArrayList<>();
dividedTasks.add(new CustomRecursiveTask(
Arrays.copyOfRange(arr, 0, arr.length / 2)));
dividedTasks.add(new CustomRecursiveTask(
Arrays.copyOfRange(arr, arr.length / 2, arr.length)));
return dividedTasks;
}
private Integer processing(int[] arr) {
return Arrays.stream(arr)
.filter(a -> a > 10 && a < 27)
.map(a -> a * 10)
.sum();
}
}Copy在此示例中，我们使用存储在CustomRecursiveTask类的arr字段中的数组来表示工作。createSubtasks（） 方法递归地将任务划分为较小的工作片段，直到每个片段都小于阈值。然后invokeAll（） 方法将子任务提交到公共池并返回Future 列表。
为了触发执行，需要为每个子任务调用join（） 方法。
我们在这里使用 Java 8 的Stream API 完成了这一点。我们使用sum（） 方法作为将子结果组合到最终结果中的表示。
4. 向ForkJoinPool提交任务
我们可以使用几种方法将任务提交到线程池。
让我们从submit（） 或execute（） 方法开始（它们的用例是相同的）：
forkJoinPool.execute(customRecursiveTask);
int result = customRecursiveTask.join();Copyinvoke（） 方法fork任务并等待结果，并且不需要任何手动连接：
int result = forkJoinPool.invoke(customRecursiveTask);CopyinvokeAll（） 方法是将一系列 ForkJoinTasks提交到ForkJoinPool 的最便捷方法。它将任务作为参数（两个任务，var args 或一个集合），fork，然后按照它们的生成顺序返回Future对象的集合。
或者，我们可以使用单独的fork（）和join（）方法。fork（） 方法将任务提交到池，但它不会触发其执行。为此，我们必须使用join（） 方法。
在递归操作的情况下，join（） 只返回null;对于递归任务<V>，它返回任务执行的结果：
customRecursiveTaskFirst.fork();
result = customRecursiveTaskLast.join();Copy在这里，我们使用invokeAll（） 方法向池提交一系列子任务。我们可以用fork（） 和join（） 做同样的工作，尽管这会对结果的排序产生影响。
为了避免混淆，通常最好使用invokeAll（） 方法将多个任务提交到ForkJoinPool。
5. 结论
使用 fork/join 框架可以加快大型任务的处理速度，但要实现此结果，我们应该遵循一些准则：
- 使用尽可能少的线程池。在大多数情况下，最好的决策是为每个应用程序或系统使用一个线程池。
- 如果不需要特定调优，请使用默认的公共线程池。
- 使用合理的阈值将ForkJoinTask拆分为子任务。
- 避免在 ForkJoinTasks 中出现任何阻塞。
腾讯云开发者
