对数组的每个元素进行并行递归函数调用（使用 ForkJoinPool）

Question

考虑以下场景：

public void processArray(int a, SomeType array) {
    for (int i = 0; i < array.length; ++i) {
        recFunction(a, array[i]);
    }
}
private void recFunction(int a, SomeType element){
     ...
     recFunction(a + x, element);
     recFunction(a + y, element);
     ...
}

processArray 对数组的每个元素调用 recFunction。如何使用 Java 创建该程序的并行版本？请注意，要处理的数组可能非常大（最多 10,000 个元素）。我的第一个想法是使用 ForkJoinPool，但我绝对不能为每个数组元素创建一个 RecursiveTask，因为那样会创建 10,000 个新对象。

另外，我必须处理不同的数组，因此必须调用 processArray 几次。我想避免每次都创建新线程并使用现有线程。有什么想法可以使用 Executor 或 ForkJoinPool 来实现吗？

Answer 1

这是一个快速排序示例，无论如何都不是最佳的，但应该为您提供与递归版本进行比较的基础。

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

public class ParallelSort {
    // default, Creates a ForkJoinPool with parallelism equal to Runtime.availableProcessors()
    private ForkJoinPool pool = new ForkJoinPool();

    public void sort(int[] elements) {
        pool.invoke(new SortAction(elements, 0, elements.length-1));
    }

    public class SortAction extends RecursiveAction {
        private static final long serialVersionUID = 3060427288506544983L;

        final int[] elements;
        int low;
        int high;

        SortAction(int[] elements, int low, int high) {
            this.elements = elements;
            this.low = low;
            this.high = high;
        }

        protected void compute() {
            if (low < high) {
                int pivotIndex = (low + high) >>> 1;
                pivotIndex = partition(pivotIndex);

                invokeAll(new SortAction(elements, low, pivotIndex - 1),
                          new SortAction(elements, pivotIndex + 1, high));
            }
        }

        private int partition(int pivotIndex) {
            int pivotValue = elements[pivotIndex];

            swap(elements, pivotIndex, high);
            int storeIndex = low;

            for (int i = low; i < high; i++) {
                if (elements[i] < pivotValue) {
                    swap(elements, i, storeIndex);
                    storeIndex++;
                }
            }

            swap(elements, storeIndex, high);
            return storeIndex;
        }

        private void swap(int[] elements, int i, int j) {
            if (i != j) {
                int temp = elements[i];
                elements[i] = elements[j];
                elements[j] = temp;
            }
        }
    }
}

一些快速说明：

• 您不需要在排序类中声明 ForkJoinPool，我只是在这里做了，所以调用者不必考虑池。

• 如果分区大小很小，这种快速排序会回退到串行排序，效果会更好。我没有在这里打扰。

这在数百万个元素上运行没有问题。