并行合并排序时出现内存不足错误

Question

我尝试并行化我的合并排序实现：http://pastebin.com/2uMGjTxr。 我想创建 Java-VM 可以提供的尽可能多的线程。我想使用java.lang.Runtime确定可能的最大线程数。

所以我想出了一个名为 MergeThread 的类：

public class MergeThread implements Runnable{

    public int[] list;
    int sIndex, eIndex;

    public MergeThread(int[] pArray, int pStartIndex, int pEndIndex){
        list = pArray;
        sIndex = pStartIndex;
        eIndex = pEndIndex;
    }

    public void run(){
        list = mergeSort(list, sIndex, eIndex);
    }

    /**
     * Merges two sorted int array into one new sorted array.
     * @param lhs
     * @param rhs
     * @return
     */
    private static int[] merge(int[] lhs, int[] rhs) {
        int[] result = new int[lhs.length + rhs.length];

        int leftIndex = 0;
        int rightIndex = 0;
        while(leftIndex < lhs.length && rightIndex < rhs.length) {
            if(lhs[leftIndex] <= rhs[rightIndex]) {
                result[leftIndex + rightIndex] = lhs[leftIndex];
                leftIndex++;
            } else {
                result[leftIndex + rightIndex] = rhs[rightIndex];
                rightIndex++;
            }
        }

        while(leftIndex < lhs.length) {
            result[leftIndex + rightIndex] = lhs[leftIndex];
            leftIndex++;
        }

        while(rightIndex < rhs.length) {
            result[leftIndex + rightIndex] = rhs[rightIndex];
            rightIndex++;
        }

        return result;
    }

    /**
     * Sorts an array from index <code>startIndex</code> (inclusive) to <code>endIndex</code> (exclusive).
     * @param array
     * @param startIndex
     * @param endIndex
     * @return new array that is sorted
     */
    private static int[] mergeSort(int[] array, int startIndex, int endIndex) {
        int length = endIndex - startIndex;
        if(length == 0) {
            return new int[]{};
        }
        if(length == 1) {
            return new int[]{array[startIndex]};
        }

        int halfLength = length / 2;
        //int[] sortedLeftPart = mergeSort(array, startIndex, startIndex + halfLength);
        MergeThread m1 = new MergeThread(array, startIndex, startIndex + halfLength);
        Thread t1 = new Thread(m1);
        t1.start();
        //int[] sortedRightPart = mergeSort(array, startIndex + halfLength, endIndex);
        MergeThread m2 = new MergeThread(array, startIndex + halfLength, endIndex);
        Thread t2 = new Thread(m2);
        t2.start();
        try{
        t1.join();
        t2.join();
        }catch(InterruptedException e){}
        return merge(m1.list, m2.list);     
    }
}

以及一个实际启动进程的类

import java.util.Random;

public class Aufg2 {
    public static Random random = new Random(100);

    public static void main(String[] args) {
        int[] array = createRandomArray(10000000);

        long time = System.currentTimeMillis();

        int[] sortedArray = sort(array);

        if(sortedArray.length != array.length || !isSorted(sortedArray)) {
            System.err.println("Failed to sort given array! :-(");
            return;
        }       
        System.out.println("Success! Sorting took " + (System.currentTimeMillis() - time) + "ms.");     
    }

    /**
     * Creates a randomly filled array of given length
     * @param length
     * @return
     */
    private static int[] createRandomArray(int length) {
        int[] result = new int[length];
        for(int i = 0; i < length; i++) {
            result[i] = random.nextInt();
        }
        return result;
    }

    /**
     * Checks whether a given int array is sorted in ascending order  
     * @param array
     * @return <code>true</code> if the given int array is sorted; <code>false</code> otherwise.
     */
    private static boolean isSorted(int[] array) {
        for(int i = 1; i < array.length; i++) {
            if(array[i] < array[i-1]) {
                return false;
            }
        }
        return true;
    }   

    /**
     * Sorts a given array (ascending order)
     * @param array
     * @return
     */
    private static int[] sort(int[] array){
        //TODO: use multiple threads to speed up the sorting
        MergeThread m = new MergeThread(array, 0, array.length);

        try{

        Thread t1 = new Thread(m);
        t1.start();
        t1.join();
        }catch(InterruptedException e){

        }
        return m.list;
    }
}

但是这种合并排序不起作用。控制台打印出很多java.lang.OutOfMemmoryError's无法创建新的本机线程。

后来消息更改为类似 java heap 的内容。

我需要更改什么才能使合并排序正常工作以及如何使用 java.lang.Runtime 来实现这一点？

Answer 1

分而治之的机制让你尝试创建类似 5000000 个线程的东西 - 并且每个线程都需要默认的 256KB (IIRC) 堆栈内存。仍然对为什么会收到 OutOfMemmoryError 感到惊讶吗？

使用 固定大小线程池 - 对池中的线程数量进行一些实验，但是任何超过系统中核心数量的东西都不太可能提高性能（并且确实可能​​会降低性能）。

Answer 2

首先，使用 ExecutorService 并在其中排队新任务，而不是创建数百万个线程（这应该消除第一个问题；如果创建数百万个线程，迟早会耗尽资源）。 1.5 倍的核心数量通常是一个不错的猜测（通常比使用可用核心数量给出更好的结果 - 但这是您必须考虑的事情）。

然后 - 如果您希望该算法在任何地方都具有高性能，那么绝对重要 - 在合理的阈值下对叶情况使用快速排序，或者如果您想要较低的阈值则使用插入排序（如果您使用插入排序，叶节点大小应该为 16 左右）工作正常）。

Answer 3

让一个线程处理数组的后半部分，而调用线程处理前半部分，

    int halfLength = length / 2;
    MergeThread m2 = new MergeThread(array, startIndex + halfLength, endIndex);
    Thread t2 = new Thread(m2);
    t2.start();//let new thread handle the second half
    array = mergeSort(array, startIndex, startIndex + halfLength);//do first half ourselves
    try{
    t2.join();
    }catch(InterruptedException e){}
    return merge(array, m2.list);

这会将创建的线程数量减少一半

，但快速排序更适合并行化，因为它不需要后递归步骤它允许线程（带有执行器的可运行作业）在委托

调用者之后立即返回，然后只需要监视所有作业何时完成