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

Question

我正在尝试使用 Java 中的并行化算法。我从合并排序开始，并在question 中发布了我的尝试。我修改后的尝试在下面的代码中，我现在尝试并行化快速排序。

在我的多线程实现或解决此问题的方法中是否存在任何新手错误？如果不是，我不应该期望双核上的顺序算法和并行算法之间的速度增加超过 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 个。

您没有获得完全双倍性能提升的原因可能是您的快速排序无法完全并行化。请注意，对快速排序的第一次调用将在初始线程开始真正并行运行之前遍历整个数组。当分流到单独的线程时，以上下文切换和模式转换的形式并行化算法也会产生开销。

看看 Fork/Join 框架，这个问题可能很适合那里。

关于实施的几点。实现 Runnable 而不是扩展 Thread。仅当您创建一些新版本的 Thread 类时才应使用扩展 Thread。当您只想做一些并行运行的工作时，最好使用 Runnable。在实现 Runnable 的同时，您还可以扩展另一个类，这为您在 OO 设计中提供了更大的灵活性。使用受限于系统中可用线程数的线程池。也不要使用 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();

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

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

Answer 2

这使用了快速排序和合并排序的组合。

import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ParallelSortMain {
    public static void main(String... args) throws InterruptedException {
        Random rand = new Random();
        final int[] values = new int[100*1024*1024];
        for (int i = 0; i < values.length; i++)
            values[i] = rand.nextInt();

        int threads = Runtime.getRuntime().availableProcessors();
        ExecutorService es = Executors.newFixedThreadPool(threads);
        int blockSize = (values.length + threads - 1) / threads;
        for (int i = 0; i < values.length; i += blockSize) {
            final int min = i;
            final int max = Math.min(min + blockSize, values.length);
            es.submit(new Runnable() {
                @Override
                public void run() {
                    Arrays.sort(values, min, max);
                }
            });
        }
        es.shutdown();
        es.awaitTermination(10, TimeUnit.MINUTES);
        for (int blockSize2 = blockSize; blockSize2 < values.length / 2; blockSize2 *= 2) {
            for (int i = 0; i < values.length; i += blockSize2) {
                final int min = i;
                final int mid = Math.min(min + blockSize2, values.length);
                final int max = Math.min(min + blockSize2 * 2, values.length);
                mergeSort(values, min, mid, max);
            }
        }
    }

    private static boolean mergeSort(int[] values, int left, int mid, int end) {
        int[] results = new int[end - left];
        int l = left, r = mid, m = 0;
        for (; l < left && r < mid; m++) {
            int lv = values[l];
            int rv = values[r];
            if (lv < rv) {
                results[m] = lv;
                l++;
            } else {
                results[m] = rv;
                r++;
            }
        }
        while (l < mid)
            results[m++] = values[l++];
        while (r < end)
            results[m++] = values[r++];
        System.arraycopy(results, 0, values, left, results.length);
        return false;
    }
}

Answer 3

如果我正确理解您的代码，请提供几个 cmets：

1. 我没有看到 numthreads 对象周围的锁，即使它可以通过多个线程访问。也许您应该将其设为 AtomicInteger。

2. 使用线程池并安排任务，即一次调用快速排序，以利用线程池。使用期货。

您当前按照您的方式划分事物的方法可能会导致较小的分区有线程，而较大的分区没有线程。也就是说，它不会优先考虑拥有自己线程的较大段。