ForkJoinPool是一个强劲的Java类,用于处理计算密集型任务。使用ForkJoinPool分解计算密集型任务并并行执行它们以获得更好的Java应用程序性能。
它的工作原理是将任务分解为更小的子任务,然后并行执行它们。该线程池使用分而治之的策略运行,使其能够并发执行任务,从而提高吞吐量并减少处理时间。
它的独特功能之一ForkJoinPool是它用于优化性能的工作窃取算法。当工作线程完成分配给它的任务时,它会从其他线程窃取任务,确保所有线程高效工作,不浪费计算机资源。
ForkJoinPool在Java的并行流和CompletableFutures中广泛使用,允许开发人员轻松并发执行任务。此外,Kotlin和Akka等其他JVM语言使用此框架来构建需要高并发性和弹性的消息驱动应用程序。
ForkJoinPool类存储workers,它们是机器上每个CPU核心上运行的进程。这些进程中的每一个都存储在Deque的双端队列中。一旦工作线程用完任务,它就会开始从其它工作线程窃取任务。
第一,会有fork任务的过程,这意味着一个大任务将被分解成可以并行执行的小任务。所有子任务完成后,它们将重新加入。然后ForkJoinPool类提供一个结果。

ForkJoinPool fork task
当一个任务被提交ForkJoinPool时,进程会被分成更小的进程并推送到一个共享队列中。一旦调用fork()方法,就会并行调用任务,直到基本条件为真。一旦处理被分叉,join()方法确保线程相互等待,直到进程完成。
所有任务最初都会被提交到一个主队列,这个主队列将任务推送到工作线程。请注意:任务是使用与堆栈数据结构一样的LIFO(后进先出)策略插入的。
还有一点很重大的是ForkJoinPool使用Deques来存储任务。这提供了会用LIFO(后进先出)或FIFO(先进先出)的能力,这是工作窃取算法所必须的。

ForkJoinPool Deque
工作窃取是一种有效的算法,它通过在池中所有可用线程之间平衡工作负载来实现计算机资源的高效使用。
当一个线程变的空闲时,它不会保持不活动状态,而是会尝试从其它扔在忙于分配给它们的工作的线程中窃取任务。此过程最大限度地利用计算资源,并确保没有线程负担过重而其它线程保持空闲状态。
工作窃取算法背后的关键概念是每个线程都有自己的双端队列任务,它以LIFO顺序执行。
当一个线程完成自己的任务并变的空闲时,它会尝试从另一个线程的双端队列任务的末尾“窃取”任务,遵循FIFO策略,与队列数据结构一样。这使得空闲线程可以接手等待时间最长的任务,从而减少等待时间,提高吞吐量。

ForkJoinPool 工作窃取算法
总体而言,ForkJoinPool的工作窃取算法是一个强劲的功能,可以通过确保有效利用所有可用的计算资源来显著提高并行程序的性能。
使用RecursiveAction类,需要继承它并覆盖compute()方法,然后,用我们想要实现的逻辑来创建子任务。
package com.joyce.forkjoin;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
public class RecursiveActionDemo {
public static void main(String[] args) {
ForkJoinPool forkJoinPool = new ForkJoinPool();
int[] array = {2, 4, 6, 8, 10};
DoubleNumber doubleNumberTask = new DoubleNumber(array, 0, array.length);
forkJoinPool.invoke(doubleNumberTask);
System.err.println(DoubleNumber.result);
}
}
class DoubleNumber extends RecursiveAction {
final int PROCESS_THRESHOLD = 2;
int[] array;
int beginIndex, endIndex;
static int result;
DoubleNumber(int[] array, int beginIndex, int endIndex) {
this.array = array;
this.beginIndex = beginIndex;
this.endIndex = endIndex;
}
@Override
protected void compute() {
if (endIndex - beginIndex <= PROCESS_THRESHOLD) {
for (int i = beginIndex; i < endIndex; i++) {
result += array[i] * 2;
}
} else {
int mid = (beginIndex + endIndex) / 2;
DoubleNumber leftArray = new DoubleNumber(array, beginIndex, mid);
DoubleNumber rightArray = new DoubleNumber(array, mid, endIndex);
// 递归调用计算方法
leftArray.fork();
rightArray.fork();
// 加入递归结果
leftArray.join();
rightArray.join();
}
}
}如代码所示,并行递归计算数组中每个数字的两倍,并计算结果。
记住重大的一点,RecursiveAction没有返回值,通过使用分而治之的策略来分解流程提高性能。正如代码所示,并非计算数组内每个元素的两倍,而是通过将数组分成多个部分来并行执行操作。
同样,需要注意的是,RecursiveAction用于可以有效分解为更小的子问题的任务时,超级有效。所以,RecursiveAction和ForkJoinPool用于计算密集型任务时,可以显著提高性能。否则,由于线程的创建和管理,性能反而会变差。
RecursiveTask和RecursiveAction之间的区别在于,方法中是存在返回值的。
package com.joyce.forkjoin;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
public class RecursiveTaskDemo extends RecursiveTask<Integer> {
private final List<Integer> numbers;
public RecursiveTaskDemo(List<Integer> numbers) {
this.numbers = numbers;
}
@Override
protected Integer compute() {
if (numbers.size() <= 2) {
return numbers.stream().mapToInt(e -> e).sum();
} else {
int mid = numbers.size() / 2;
List<Integer> list1 = numbers.subList(0, mid);
List<Integer> list2 = numbers.subList(mid, numbers.size());
RecursiveTaskDemo task1 = new RecursiveTaskDemo(list1);
RecursiveTaskDemo task2 = new RecursiveTaskDemo(list2);
task1.fork();
return task1.join() + task2.compute();
}
}
public static void main(String[] args) {
ForkJoinPool forkJoinPool = new ForkJoinPool();
List<Integer> numbers = Arrays.asList(1,3,5,7,9);
int output = forkJoinPool.invoke(new RecursiveTaskDemo(numbers));
System.err.println(output);
}
}
此示例代码,递归分解数组,直到达到基本条件。将list1和list2加入到RecursiveTask,然后分叉task1,利用compute()并行执行方法和数组的其它部分。当递归达到条件,join()方法加入结果。
ForkJoinPool虽然很便利,但是不应在所有的情况下使用。正如前面所讲,最好将它用于高密度计算的并发进程。
本文主要介绍了如何使用ForkJoinPool功能在CPU内核中执行繁重的操作。我们来总结一下: