Java：通过多线程并行化快速排序

Question

我正在尝试使用Java并行化算法。我从合并排序开始，然后在这个问题上发表了自己的尝试。我修改后的尝试在下面的代码中，在这里我尝试并行化快速排序。

我的多线程实现或方法中是否有菜鸟错误？如果不是，我是否应该期望对决核上的顺序算法和并行算法之间的速度提高超过32％（请参阅底部的时序）？

这是多线程算法：

    public class ThreadedQuick extends Thread
    {
        final int MAX_THREADS = Runtime.getRuntime().availableProcessors();

        CountDownLatch doneSignal;
        static int num_threads = 1;

        int[] my_array;
        int start, end;

        public ThreadedQuick(CountDownLatch doneSignal, int[] array, int start, int end) {
            this.my_array = array;
            this.start = start;
            this.end = end;
            this.doneSignal = doneSignal;
        }

        public static void reset() {
            num_threads = 1;
        }

        public void run() {
            quicksort(my_array, start, end);
            doneSignal.countDown();
            num_threads--;
        }

        public void quicksort(int[] array, int start, int end) {
            int len = end-start+1;

            if (len <= 1)
                return;

            int pivot_index = medianOfThree(array, start, end);
            int pivotValue = array[pivot_index];

            swap(array, pivot_index, end);

            int storeIndex = start;
            for (int i = start; i < end; i++) {
               if (array[i] <= pivotValue) {
                   swap(array, i, storeIndex);
                   storeIndex++;
               }
            }

            swap(array, storeIndex, end);

            if (num_threads < MAX_THREADS) {
                num_threads++;

                CountDownLatch completionSignal = new CountDownLatch(1);

                new ThreadedQuick(completionSignal, array, start, storeIndex - 1).start();
                quicksort(array, storeIndex + 1, end);

                try {
                    completionSignal.await(1000, TimeUnit.SECONDS);
                } catch(Exception ex) {
                    ex.printStackTrace();
                }
            } else {
                quicksort(array, start, storeIndex - 1);
                quicksort(array, storeIndex + 1, end);
            }
        }
    }

这是我的开始方式：

ThreadedQuick.reset();
CountDownLatch completionSignal = new CountDownLatch(1);
new ThreadedQuick(completionSignal, array, 0, array.length-1).start();
try {
    completionSignal.await(1000, TimeUnit.SECONDS);
} catch(Exception ex){
    ex.printStackTrace();
}

我针对Arrays.sort和类似的顺序快速排序算法进行了测试。以下是英特尔决斗核心戴尔笔记本电脑上的计时结果，以秒为单位：

元素：500,000，顺序：0.068592，螺纹：0.046871，Arrays.sort：0.079677

元素：1,000,000，顺序：0.14416，线程：0.095492，Arrays.sort：0.167155

元素：2,000,000，顺序：0.301666，线程：0.205719，Arrays.sort：0.350982

元素：4,000,000，顺序：0.623291，线程：0.424119，Arrays.sort：0.712698

元素：8,000,000，顺序：1.279374，线程：0.859363，Arrays.sort：1.487671

上面的每个数字是100次测试的平均时间，排除了3个最低和3个最高的案例。我使用Random.nextInt（Integer.MAX_VALUE）为每个测试生成一个数组，并使用相同的种子每10个测试对其进行一次初始化。每个测试都包括使用System.nanoTime对给定算法进行计时。平均后，我将四舍五入到小数点后六位。显然，我确实检查了每种方法是否都
有效 。

如您所见，在每组测试中，顺序案例和线程案例之间的速度大约提高了32％。就像我在上面问过的，我不应该期望更多吗？

Answer 1

使numThreads静态化可能会导致问题，在某些时候您最终可能会运行超过MAX_THREADS个。

您之所以无法获得完全双倍的性能，可能是因为您的快速排序无法完全并行化。请注意，对quicksort的第一次调用将在初始线程真正开始并行运行之前，在初始线程中对整个数组进行传递。当耕种到单独的线程时，以上下文切换和模式转换的形式并行化算法也存在开销。

看一下Fork / Join框架，这个问题可能很整洁。

关于实施的几点。实现可运行而不是扩展线程。仅当您创建一些新版本的Thread类时，才应使用扩展Thread。当您只想做一些可以并行运行的工作时，最好使用Runnable。在实现Runnable的同时，您仍然可以扩展另一个类，从而为您提供面向对象设计的更多灵活性。使用限制为您在系统中可用的线程数的线程池。也不要使用numThreads来决定是否派生新线程。您可以预先计算。使用最小分区大小，该大小是整个阵列的大小除以可用处理器的数量。就像是：

public class ThreadedQuick implements Runnable {

    public static final int MAX_THREADS = Runtime.getRuntime().availableProcessors();
    static final ExecutorService executor = Executors.newFixedThreadPool(MAX_THREADS);

    final int[] my_array;
    final int start, end;

    private final int minParitionSize;

    public ThreadedQuick(int minParitionSize, int[] array, int start, int end) {
        this.minParitionSize = minParitionSize;
        this.my_array = array;
        this.start = start;
        this.end = end;
    }

    public void run() {
        quicksort(my_array, start, end);
    }

    public void quicksort(int[] array, int start, int end) {
        int len = end - start + 1;

        if (len <= 1)
            return;

        int pivot_index = medianOfThree(array, start, end);
        int pivotValue = array[pivot_index];

        swap(array, pivot_index, end);

        int storeIndex = start;
        for (int i = start; i < end; i++) {
            if (array[i] <= pivotValue) {
                swap(array, i, storeIndex);
                storeIndex++;
            }
        }

        swap(array, storeIndex, end);

        if (len > minParitionSize) {

            ThreadedQuick quick = new ThreadedQuick(minParitionSize, array, start, storeIndex - 1);
            Future<?> future = executor.submit(quick);
            quicksort(array, storeIndex + 1, end);

            try {
                future.get(1000, TimeUnit.SECONDS);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        } else {
            quicksort(array, start, storeIndex - 1);
            quicksort(array, storeIndex + 1, end);
        }
    }    
}

您可以通过执行以下操作开始：

ThreadedQuick quick = new ThreadedQuick(array / ThreadedQuick.MAX_THREADS, array, 0, array.length - 1);
quick.run();

这将在同一线程中启动排序，从而避免了启动时不必要的线程跳跃。

警告：不确定上面的实现实际上会更快，因为我还没有对其进行基准测试。