2

我有合并排序的示例实现,一个使用 Fork-Join,另一个是直接递归函数。

看起来 fork-join 比直接递归慢,为什么?

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

class DivideTask extends RecursiveTask<int[]> {
    private static final long serialVersionUID = -7017440434091885703L;
    int[] arrayToDivide;

    public DivideTask(int[] arrayToDivide) {
        this.arrayToDivide = arrayToDivide;
    }

    @Override
    protected int[] compute() {
        //List<RecursiveTask> forkedTasks = new ArrayList<>();

        /*
         * We divide the array till it has only 1 element. 
         * We can also custom define this value to say some 
         * 5 elements. In which case the return would be
         * Arrays.sort(arrayToDivide) instead.
         */
        if (arrayToDivide.length > 1) {

            List<int[]> partitionedArray = partitionArray();

            DivideTask task1 = new DivideTask(partitionedArray.get(0));
            DivideTask task2 = new DivideTask(partitionedArray.get(1));
            invokeAll(task1, task2);

            //Wait for results from both the tasks
            int[] array1 = task1.join();
            int[] array2 = task2.join();

            //Initialize a merged array
            int[] mergedArray = new int[array1.length + array2.length];

            mergeArrays(task1.join(), task2.join(), mergedArray);

            return mergedArray;
        }
        return arrayToDivide;
    }

    private void mergeArrays(int[] array1, int[] array2, int[] mergedArray) {

        int i = 0, j = 0, k = 0;

        while ((i < array1.length) && (j < array2.length)) {

            if (array1[i] < array2[j]) {
                mergedArray[k] = array1[i++];
            } else {
                mergedArray[k] = array2[j++];
            }

            k++;
        }

        if (i == array1.length) {
            for (int a = j; a < array2.length; a++) {
                mergedArray[k++] = array2[a];
            }
        } else {
            for (int a = i; a < array1.length; a++) {
                mergedArray[k++] = array1[a];
            }
        }
    }

    private List<int[]> partitionArray() {
        int[] partition1 = Arrays.copyOfRange(arrayToDivide, 0, arrayToDivide.length / 2);

        int[] partition2 = Arrays.copyOfRange(arrayToDivide, arrayToDivide.length / 2, arrayToDivide.length);
        return Arrays.asList(partition1, partition2);
    }
}

public class ForkJoinTest {
    static int[] numbers;
    static final int SIZE = 1_000_000;
    static final int MAX = 20;

    public static void main(String[] args) {
        setUp();

        testMergeSortByFJ();
        testMergeSort();
    }

    static void setUp() {
        numbers = new int[SIZE];
        Random generator = new Random();
        for (int i = 0; i < numbers.length; i++) {
            numbers[i] = generator.nextInt(MAX);
        }
    }

    static void testMergeSort() {
        long startTime = System.currentTimeMillis();

        Mergesort sorter = new Mergesort();
        sorter.sort(numbers);

        long stopTime = System.currentTimeMillis();
        long elapsedTime = stopTime - startTime;
        System.out.println("Mergesort Time:" + elapsedTime + " msec");
    }

    static void testMergeSortByFJ() {
        //System.out.println("Unsorted array: " + Arrays.toString(numbers));
        long t1 = System.currentTimeMillis();
        DivideTask task = new DivideTask(numbers);
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        forkJoinPool.invoke(task);
        //System.out.println("Sorted array: " + Arrays.toString(task.join()));
        System.out.println("Fork-Join Time:" + (System.currentTimeMillis() - t1) + " msec");
    }
 }

class Mergesort {
    private int[] msNumbers;
    private int[] helper;

    private int number;

    private void merge(int low, int middle, int high) {

        // Copy both parts into the helper array
        for (int i = low; i <= high; i++) {
            helper[i] = msNumbers[i];
        }

        int i = low;
        int j = middle + 1;
        int k = low;
        // Copy the smallest values from either the left or the right side back
        // to the original array
        while (i <= middle && j <= high) {
            if (helper[i] <= helper[j]) {
                msNumbers[k] = helper[i];
                i++;
            } else {
                msNumbers[k] = helper[j];
                j++;
            }
            k++;
        }
        // Copy the rest of the left side of the array into the target array
        while (i <= middle) {
            msNumbers[k] = helper[i];
            k++;
            i++;
        }

    }

    private void mergesort(int low, int high) {
        // Check if low is smaller then high, if not then the array is sorted
        if (low < high) {
            // Get the index of the element which is in the middle
            int middle = low + (high - low) / 2;
            // Sort the left side of the array
            mergesort(low, middle);
            // Sort the right side of the array
            mergesort(middle + 1, high);
            // Combine them both
            merge(low, middle, high);
        }
    }

    public void sort(int[] values) {
        this.msNumbers = values;
        number = values.length;
        this.helper = new int[number];
        mergesort(0, number - 1);
    }
}
4

4 回答 4

5

恕我直言,主要原因不是线程产生和池化造成的开销。

我认为多线程版本运行缓慢主要是因为您一直在不断地创建新数组数百万次。最终,您创建了 100 万个包含单个元素的数组,这让垃圾收集器很头疼。

您所有的DivideTasks 都可以对数组的不同部分(两半)进行操作,因此只需向它们发送一个范围并让它们在该范围内操作。

此外,您的并行化策略使得无法使用巧妙的“辅助数组”优化(注意helper顺序版本中的数组)。这种优化将“输入”数组与在其上进行合并的“辅助”数组交换,因此不应该为每个合并操作创建一个新数组:这是一种节省内存的技术,如果你不这样做,你就无法做到t按递归树的级别并行化。

对于课堂作业,我必须并行化 MergeSort,并且通过按递归树的级别并行化,我设法获得了不错的加速。不幸的是,代码是 C 语言并使用 OpenMP。如果你要我可以提供。

于 2013-01-03T00:44:04.743 回答
4

正如 gd1 指出的那样,您正在进行大量的数组分配和复制;这会让你付出代价。相反,您应该在同一个数组的不同部分上工作,注意没有子任务在另一个子任务正在处理的部分上工作。

但除此之外,fork/join 方法(与任何并发性一样)会带来一定的开销。事实上,如果您查看RecursiveTask的 javadocs ,他们甚至指出他们的简单示例执行缓慢,因为分叉过于细化。

长话短说,你应该有更少的细分,每个细分都做得更多。更一般地说,任何时候你的非阻塞线程多于核心,吞吐量都不会提高,实际上开销会开始减少。

于 2013-01-03T00:51:56.957 回答
0

如果不深入查看您的代码,则生成一个新线程的成本很高。如果您没有太多工作要做,那么仅出于性能原因就不值得。在这里很笼统地说,但是在产生新线程并开始运行之前,单个线程可能会循环数千次(尤其是在 Windows 上)。

请参阅Doug Lea 的论文(在 2. DESIGN 下),其中他指出:

“然而,java.lang.Thread 类(以及 Java 线程通常基于的 POSIX pthreads)是支持 fork/join 程序的次优工具”

于 2013-01-03T00:38:54.290 回答
0

还发现以下有关使用 Fork/Join 的信息Dan Grossman 的 Fork/Join 简介

于 2013-01-03T23:36:18.570 回答