排序算法之多线程实现

单线程归并排序算法

在排序算法中，归并排序(Merge Sort)采用了经典的分治(divide-and-conquer)策略，并且具有并行的特征。如下是归并排序算法串行执行的Java代码：

public class MergeSort {
    /**
     * 将给定数组排序(单线程，归并排序)
     * @param arr int整型一维数组 待排序的数组
     * @return int整型一维数组 排序后的数组
     */
    public static int[] sort(final int[] arr) {
        if (arr == null) {
            return null;
        }
        return sort(arr, 0, arr.length - 1);
    }

    private static int[] sort(final int[] arr, int start, int end) {
        if (start > end) {
            return new int[]{};
        }
        if (start == end) {
            return new int[]{arr[start]};
        }
        int mid = (end - start) / 2 + start;
        int[] left = sort(arr, start, mid);
        int[] right = sort(arr, mid + 1, end);
        return merge(left, right);
    }

    private static int[] merge(int[] left, int[] right) {
        int lenL = left.length;
        int lenR = right.length;
        int len = lenL + lenR;
        int[] res = new int[len];
        int i = 0;
        int j = 0;
        int k = 0;
        while (i < lenL && j < lenR) {
            if (left[i] < right[j]) {
                res[k++] = left[i++];
            } else {
                res[k++] = right[j++];
            }
        }
        while (i < lenL) {
            res[k++] = left[i++];
        }
        while (j < lenR) {
            res[k++] = right[j++];
        }
        return res;
    }
}

多线程排序算法

从归并排序算法串行执行的Java代码中可以看出，归并排序的核心在于，将要排序的数组分为左半部分和右半部分分别进行排序，再对排序后的这两部分进行合并。由于对左半部分的排序和对右半部分的排序互不干扰，因此可以并行计算。在多核处理器系统中设置两个线程分别处理对左半部分的排序和对右半部分的排序可以大大提到归并排序的效率。

需要注意的是：

1. 当左右两半部分数据很小时，创建线程所需要耗费的资源将远远高于排序所需要耗费的资源。因此在实现的过程中，当左右两半部分数据较小时，应该使用串行排序而不在创建新的线程。
2. 应该严格控制线程的数量，过多的线程会占用大量内存，对线程的管理以及线程间的切换都会消耗CUP资源。

如下是多线程排序算法Java代码：

import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MergeSortWithMultithread {
    /**
     * 将给定数组排序(多线程，归并排序)
     * @param arr int整型一维数组 待排序的数组
     * @return int整型一维数组 排序后的数组
     */

    private static ExecutorService pool;

    public static int[] sort(final int[] arr) {
        if (arr == null) {
            return null;
        }
        if (pool == null || pool.isShutdown()) {
            pool = Executors.newCachedThreadPool();
        }
        int[] res = sort(arr, 0, arr.length - 1, 0);
        pool.shutdown();
        pool = null;
        return res;
    }

    private static int[] sort(final int[] arr, int start, int end, int k) {
        if (start > end) {
            return new int[]{};
        }
        if (start == end) {
            return new int[]{arr[start]};
        }
        if ((k >= 6) || (end - start <= 1000)) {
            int[] res = Arrays.copyOfRange(arr, start, end + 1);
            Arrays.sort(res);
            return res;
        }
        int mid = (end - start) / 2 + start;
        CompletableFuture<int[]> left = CompletableFuture.supplyAsync(() -> {
            return sort(arr, start, mid, k + 1);
        }, pool);
        CompletableFuture<int[]> right = CompletableFuture.supplyAsync(() -> {
            return sort(arr, mid + 1, end, k + 1);
        }, pool);
        CompletableFuture<int[]> res = left.thenCombineAsync(right, (a, b) -> {
            return merge(a, b);
        }, pool);
        return res.join();
    }

    private static int[] merge(int[] left, int[] right) {
        int lenL = left.length;
        int lenR = right.length;
        int len = lenL + lenR;
        int[] res = new int[len];
        int i = 0;
        int j = 0;
        int k = 0;
        while (i < lenL && j < lenR) {
            if (left[i] < right[j]) {
                res[k++] = left[i++];
            } else {
                res[k++] = right[j++];
            }
        }
        while (i < lenL) {
            res[k++] = left[i++];
        }
        while (j < lenR) {
            res[k++] = right[j++];
        }
        return res;
    }
}

实验代码及实验结果

如下是实验Java代码：

import java.util.Arrays;
import java.util.Random;

public class Main {
    public static void main(String[] args) throws Exception {
        // 生成测试数据
        final int len = (int) 3e7;
        final int[] arr = new int[len];
        Random r = new Random(System.currentTimeMillis());
        for (int i = 0; i < len; ++i) {
            arr[i] = r.nextInt();
        }

        //定义用于测试运行时间变量
        long startTime = 0;
        long endTime = 0;

        // 测试Java自带的排序函数
        startTime = System.currentTimeMillis();
        int[] sortedArr1 = Arrays.copyOf(arr, arr.length);
        Arrays.sort(sortedArr1);
        endTime = System.currentTimeMillis();
        System.out.println("Java自带的排序函数用时：" + (endTime - startTime));

        // 测试单线程的归并排序算法
        startTime = System.currentTimeMillis();
        int[] sortedArr2 = MergeSort.sort(arr);
        endTime = System.currentTimeMillis();
        System.out.println("单线程的归并排序算法用时：" + (endTime - startTime));

        // 测试多线程的排序算法
        startTime = System.currentTimeMillis();
        int[] sortedArr3 = MergeSortWithMultithread.sort(arr);
        endTime = System.currentTimeMillis();
        System.out.println("多线程的排序算法用时：" + (endTime - startTime));

        // 检查三种方案的排序结果是否一致
        if (Arrays.equals(sortedArr1, sortedArr2) && Arrays.equals(sortedArr1, sortedArr3)) {
            System.out.println("三种方案的排序结果一致");
        } else {
            System.out.println("三种方案的排序结果不一致");
        }
    }
}

实验结果如下：

扩展

归并排序中使用Fork/Join框架与普通多线程的比较