ForkJoinPool的了解与

ForkJoinPool是一个强大的Java类，用于处理计算密集型任务。使用ForkJoinPool分解计算密集型任务并并行执行它们以获得更好的Java应用程序性能。

它的工作原理是将任务分解为更小的子任务，然后并行执行它们。该线程池使用分而治之的策略运行，使其能够并发执行任务，从而提高吞吐量并减少处理时间。

它的独特功能之一ForkJoinPool是它用于优化性能的工作窃取算法。当工作线程完成分配给它的任务时，它会从其他线程窃取任务，确保所有线程高效工作，不浪费计算机资源。

ForkJoinPool在Java的并行流和CompletableFutures中广泛使用，允许开发人员轻松并发执行任务。此外，Kotlin和Akka等其他JVM语言使用此框架来构建需要高并发性和弹性的消息驱动应用程序。

ForkJoinPool解析

ForkJoinPool类存储workers，它们是机器上每个CPU核心上运行的进程。这些进程中的每一个都存储在Deque的双端队列中。一旦工作线程用完任务，它就会开始从其它工作线程窃取任务。

首先，会有fork任务的过程，这意味着一个大任务将被分解成可以并行执行的小任务。所有子任务完成后，它们将重新加入。然后ForkJoinPool类提供一个结果。

ForkJoinPool fork task

当一个任务被提交ForkJoinPool时，进程会被分成更小的进程并推送到一个共享队列中。一旦调用fork()方法，就会并行调用任务，直到基本条件为真。一旦处理被分叉，join()方法确保线程相互等待，直到进程完成。

所有任务最初都会被提交到一个主队列，这个主队列将任务推送到工作线程。请注意：任务是使用与堆栈数据结构相同的LIFO（后进先出）策略插入的。

还有一点很重要的是ForkJoinPool使用Deques来存储任务。这提供了会用LIFO（后进先出）或FIFO（先进先出）的能力，这是工作窃取算法所必须的。

ForkJoinPool Deque

工作窃取算法

工作窃取是一种有效的算法，它通过在池中所有可用线程之间平衡工作负载来实现计算机资源的高效使用。

当一个线程变的空闲时，它不会保持不活动状态，而是会尝试从其它扔在忙于分配给它们的工作的线程中窃取任务。此过程最大限度地利用计算资源，并确保没有线程负担过重而其它线程保持空闲状态。

工作窃取算法背后的关键概念是每个线程都有自己的双端队列任务，它以LIFO顺序执行。

当一个线程完成自己的任务并变的空闲时，它会尝试从另一个线程的双端队列任务的末尾“窃取”任务，遵循FIFO策略，与队列数据结构相同。这使得空闲线程可以接手等待时间最长的任务，从而减少等待时间，提高吞吐量。

ForkJoinPool 工作窃取算法

总体而言，ForkJoinPool的工作窃取算法是一个强大的功能，可以通过确保有效利用所有可用的计算资源来显著提高并行程序的性能。

ForkJoinPool的主要类

• ForkJoinPool，创建线程池使用ForkJoin框架，它的工作方式与其它线程池类似。此类中最重要的方法是commonPool()方法来创建ForkJoin线程池。
• RecursiveAction，该类的主要功能是计算递归。在compute()方法是没有返回值的。
• RecursiveTask，此类的工作方式和RecursiveAction类似，不同之处在于compute()方法是有返回值的。

使用

RecursiveAction

使用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

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虽然很便利，但是不应在所有的情况下使用。正如前面所讲，最好将它用于高密度计算的并发进程。

• 递归任务：ForkJoinPool非常适合执行递归算法，例如快速排序、归并排序或者二分查找。这些算法可以分解为更小的子问题并行执行，从而显著提高性能。
• 高并发场景：在高并发场景中，比如数据处理、Web服务器等，可以使用跨线程并行执行ForkJoinPool任务，有助于提高性能和吞吐量。
• 并行问题：如果你有一个可以很容易地分成独立子任务的问题，比如说图像处理或者数值模拟等，可以使用ForkJoinPool并行执行子任务。

总结

本文主要介绍了如何使用ForkJoinPool功能在CPU内核中执行繁重的操作。我们来总结一下：

• ForkJoinPool是一个线程池，使用分而治之的策略来递归执行任务。
• ForkJoinPool并行执行任务，从而有效利用计算机资源，提升性能。
• 工作窃取算法通过允许空闲线程从繁忙线程窃取任务来优化资源利用率。
• 任务存储在双端队列中，存储采用LIFO策略，窃取采用FIFO策略。
• 框架中的主要类，ForkJoinPool、RecursiveAction、RecursiveTask：ForkJoinPool通常与并行流和CompletableFuture。RecursiveAction用于计算递归操作且没有返回值。RecursiveTask与RecursiveAction类似，但有返回值。compute()方法在两个类中都被重写以实现自定义逻辑。fork()方法调用compute()方法并将任务分解为更小的子任务。join()方法等待子任务完成并合并结果。