排序算法（九）- Java源码中的DualPivotQuicksort

前言

上篇文章中我们了解了双轴快排和三路快排的原理及实现，排序算法（七） - 双轴快速排序和排序算法（八） - 三路快速排序，了解了虽然快速排序有着较高的效率，但是在数据坏的情况下排序速度非常不乐观。

而后提到了Java源码中的DualPivotQuicksort，它是一种混合排序算法。

而双轴快排和三路快排正是Java源码中的DualPivotQuicksort的一部分，我们来看下它吧。

正文

算法原理

这个排序算法是Vladimir Yaroslavskiy、Jon Bentley、Josh Bloch在2011年完成的，最早见于JDK1.7版本。

它是一种混合排序算法，其内部主要包括插入排序、三路快排、双轴快排、计数排序、归并排序这几种排序算法，将几种排序算法的优点发挥出来。

在了解排序原理之前，先了解几个固定值，这几个固定值是根据大量实验确定的数据。

• MAX_RUN_COUNT = 67

  这个值，可以用来指使用归并排序的最大RUN数量，也可以指校验数据是否可以使用归并排序的一个标准。

  比如对于int型数据int[] array，DualPivotQuicksort在开始时会从数组头部开始，寻找连续有序数据段，寻找到一段，就把末尾下标记录进一个run，如果该段逆序，就把它倒过来，比如[1,2,8,5,3,2,7,1,9]，它的有序段分隔为[1,2,8][5,3,2][7,1][9]，其中[5,3,2]和[7,1]在处理过程中会倒过来，这样如果当数据量大时，明显run数量会超过67，如果超了，就直接使用优化的快排处理了。

  PS：上面的例子我们可以看到最后的run为

  run[0] = 2 = index0 -index 2= [1,2,8] ;

  run[1] = 5 = index3 - index5 = [2,3,5] ;

  run[2] = 7 = index6 -index7= [1,7] ;

  run[3] = 8 = index8 = [9] ;

  其实run[2]和run[3]是可以合并的，这样可以节省run数量。合并条件就是后一个run的第一个元素要大于等于前一个run的最后一个元素。JDK1.8并未对这块做处理，JDK9及以上版本这块有了优化：

  if (run[count] > left && a[run[count]] >= a[run[count] - 1]) {
    count--; 
  }

  会处理这种情况。

• QUICKSORT_THRESHOLD = 286

  这个值表示快排阈值，如果数据量不超过286，就直接使用优化的快排处理。

• INSERTION_SORT_THRESHOLD = 47

  这个值表示插入排序阈值，对于长度小于等于这个数的数据，就使用插入排序（或变体）处理。

• COUNTING_SORT_THRESHOLD_FOR_BYTE = 29

  对于byte类型数据，如果数据长度小于等于这个值，直接使用插入排序处理。（超过会使用计数排序处理）

• COUNTING_SORT_THRESHOLD_FOR_SHORT_OR_CHAR = 3200

  对于short或者char类型数据，如果长度超过3200阈值，就会使用计数排序处理。

• NUM_SHORT_VALUES = 1 << 16 = 65536

  对于short类型数据，使用计数排序时，初始化桶的数量。（-32768 <= short <= 32767）

• NUM_CHAR_VALUES = 1 << 16 = 65536

  对于char类型数据，使用计数排序时，初始化桶的数量。（字符集范围Unicode 0-65535）

• NUM_BYTE_VALUES = 1 << 8 = 256

  对于byte类型数据，使用计数排序，初始化桶的数量。（-128 <= byte <= 127）

以上前五个都是大量实验确定的值。

DualPivotQuicksort的算法原理可以大致描述如下：

int类型数据

1. 对于待排序数组int[] array，如果数据长度小于 QUICKSORT_THRESHOLD = 286，直接使用优化的快速排序sort；

2. 否则校验数据，构建一个 MAX_RUN_COUNT+1 = 68 的数组，用来存储run，也用来校验是否可以使用归并排序或者使用优化的快排sort；

3. 对于待排序数组，依次寻找连续有序数组段，将末尾index记录进run，由于会出现相等的数据值：

   • JDK1.8的处理方式是统计连续相等值的数量，超过 MAX_RUN_LENGTH=33 时会直接使用优化后的快速排序sort；
   • JDK9及以上版本的处理方式为将这些相等值统计进了上一个run里。逆序后相邻两个run可以保持升序的也会被合并。

   如上面例子：[1,2,8,5,3,2,7,1,9] 会分成[1,2,8][5,3,2][7,1][9]，倒序的逆序得到[1,2,8][2,3,5][1,7][9] 这四个run，JDK9及以上版本[1,7][9]这两个run会合并为一个[1,7,9]，得益于这个校验。

   if (run[count] > left && a[run[count]] >= a[run[count] - 1]) {
               count--;
   }

4. 如果统计过程中run数量超过MAX_RUN_COUNT（限制条件++count == MAX_RUN_COUNT），则说明数据不是高度结构化的（是杂乱无章的），就会使用优化后的快排进行处理sort；如果是高度结构化的（部分有序的），会使用归并排序进行处理。

5. 再来说下优化后的快速排序sort：

   • 如果输入数据长度小于 INSERTION_SORT_THRESHOLD = 47 ，则直接使用插入排序（或变种）进行处理。
   • 具体使用普通的插入排序还是变种的插入排序，取决于leftmost这个boolean值，如果true的话就使用普通的插排；否则使用首部部分部分有序的插排（变种）。
   • 如果长度大于等于47，就要开始真正的快排部分了，首先它会寻找5个点index（e1,e2,e3,e4,e5），这5个index可以大致把数据分成等长的7份（其中e3近似为数据长度中点）；
   • 对这5个index上的值进行排序同时交换了它们的位置（用的是穷举法的插入排序）；
   • 如果这5个点的值都不相等，用 array[e2] 和 array[e4] 作为基准轴，对数据进行双轴快排处理，这里面有一点优化，（我们知道双轴快排会把数据分成3份，左部分、中间部分、右部分）就是如果中间部分长度超过4/7总长度，会跳过等于pivot（基准值）的元素，因为它们是相等的；
   • 如果5个点的值全部相等，就使用三路快排处理，其轴为 array[e3] 即中间值（5个值都相等，较大概率说明数据的重复度很高了）。

long类型数据

同int类型数据处理

short类型数据

1. 对于待排序数组short[] array，如果数据长度大于 COUNTING_SORT_THRESHOLD_FOR_SHORT_OR_CHAR = 3200 ，直接使用计数排序处理，计数排序的初始化桶数量为 NUM_SHORT_VALUES = 65536；

2. 否则对于小于等于3200的数据，参考int类型处理。

char类型数据

参考short类型处理方式

byte类型数据

1. 对于待排序数组byte[] array，如果数据长度大于 COUNTING_SORT_THRESHOLD_FOR_BYTE = 29 ，直接使用计数排序处理，其中初始化的桶数量为 NUM_BYTE_VALUES = 256；

2. 否则数据长度小于等于29，直接使用插入排序法处理。

float类型数据

1. 对于待排序数组float[] array，先看看数据中有没有NaN这种数据，有的话直接与尾部数据交换，排序边界位置也应该缩小，因为NaN是无法参与比较的；

2. 对于剩下可以比较大小的数据部分，直接参考int类型数据的处理方式；

3. 排好后，由于可能有正0和负数0（-0.0 和 +0.0）这种数据，这种情况下负0应该在正0前面的，需要特殊处理下（程序中使用了 Float.floatToRawIntBits方法处理，这个可以返回浮点值的实际表示形式，可以确定正负）。

double类型数据

参考float类型处理方式

以上就是DualPivotQuicksort的逻辑原理。

相关源代码

我们可以在 java.util.DualPivotQuicksort找到这个类，它是一个not public且final，且构造函数为private的类。

下图是我对它源码的一些解读，有兴趣的可以看看，这儿使用的是JDK1.8_131版本，在高版本的JDK中，某些部分有些改动，但不影响整体阅读。

这个源码有3000多行，考虑到文字数量及可读性问题，我使用了图片，方便大家阅读部分逻辑。

其它

上面我们看到Java里优化后的DualPivotQuicksort代码量是十分巨大的，这儿我们以int[]举例，来检测DualPivotQuicksort的效率问题。

我们创建几个测试数组，如下：

public class SortTest {
    /**
     * 随机生成数组
     * @param length
     * @return
     */
    private static int[] randomArray(int length){
        int[] a = new int[length];
        Random r = new Random();
        for (int i = 0; i < a.length; i++) {
            a[i] = r.nextInt(length);
        }
        return a;
    }
    /**
     * 生成一个有序的倒序数组
     * @param length
     * @return
     */
    private static int[] reversalArray(int length){
        int[] a = new int[length];
        for (int i = 0; i < a.length; i++) {
            a[i] = a.length - i - 1;
        }
        return a;
    }
    /**
     * 生成有个有序的正序数组
     * @param length
     * @return
     */
    private static int[] orderArray(int length){
        int[] a = new int[length];
        for (int i = 0; i < a.length; i++) {
            a[i] = i;
        }
        return a;
    }
    /**
     * 生成前X部分有序后length-X部分无序的数组
     * @param length
     * @return
     */
    private static int[] orderAndRandomArray(int length,int x){
        if(x >= length){
            throw new RuntimeException("x不能超过length");
        }
        int[] a = new int[length];
        Random r = new Random();
        for (int i = 0; i < x; i++) {
            a[i] = i;
        }
        for(int i=a.length - x;i<a.length;i++ ){
            a[i] = r.nextInt(length);
        }
        return a;
    }
    /**
     * 生成前x部分无序后length-x部分有序的数组
     * @param length
     * @param x
     * @return
     */
    private static int[] randomAndOrderArray(int length,int x){
        if(x >= length){
            throw new RuntimeException("x不能超过length");
        }
        int[] a = new int[length];
        Random r = new Random();
        for (int i = 0; i < x; i++) {
            a[i] = r.nextInt(length);
        }
        for(int i=a.length - x;i<a.length;i++ ){
            a[i] = i;
        }
        return a;
    }
    private static void doSort(int[] a){
        System.out.println("Java-DualPivotQuicksort排序开始：");
        long start = System.currentTimeMillis();
        JavaDualPivotQuicksort.sort(a,0,a.length-1,null,0,0);
        System.out.println("Java-DualPivotQuicksort耗时："+(System.currentTimeMillis()-start)+"ms");
        System.out.println("Java-DualPivotQuicksort排序完成！");
        System.out.println("数组是否有序："+isOrdered(a));
    }
    public static void main(String[] args) {
        final int length = 100000000;
        //随机无序数组
        int[] a1 = randomArray(length);
        doSort(a1);
        //有序数组
        int[] a2 = orderArray(length);
        doSort(a2);
        //逆序数组
        int[] a3 = reversalArray(length);
        doSort(a3);
        //前部分有序后部分无序数组
        int[] a4 = orderAndRandomArray(length,length/3);
        doSort(a4);
        //前部分无序后部分有序数组
        int[] a5 = randomAndOrderArray(length,length/3);
        doSort(a5);
        //前部分有序后部分无序数组
        int[] a6 = orderAndRandomArray(length,length - length/3);
        doSort(a6);
        //前部分无序后部分有序数组
        int[] a7 = randomAndOrderArray(length,length/2);
        doSort(a7);
    }
}

最后可以得到类似的结果：

PS：我们可以使用之前提到的所有排序算法进行测试，其效率都不如该算法稳定。对于O(n^2)复杂度的排序算法，很多出现内存或者栈溢出异常。

我们进一步分析下该算法。

可以看到该算法排序取决于几个要素：数据长度、数据类型、数据是否结构化（部分有序）。

其算法比较突出的地方有以下几点：

• 创建了 MAX_RUN_COUNT+1 个run，这个run一个重要的作用就是判断数据是否结构化，如果是的话用归并处理要快很多；如果不是，再使用快排，相比直接快排，只是浪费了一些校验时间，却规避了快排O(n^2)复杂度出现的情况，这也是run的数量要取合适的一个原因，算法作者通过大量实验确定了67这个数。

• 当对于short、char、byte类型数据时，由于其范围不是很大，计数排序的优势就体现出来了，因此使用了计数排序。

• 对数据进行快排时，首先选择了5个点，在分情况从中获取基准值，并决定使用双轴快排还是三路快排。

  这儿有一点是比较有意思的，我们可以看下源码中的这个优化后的快排算法sort(int[] a, int left, int right, boolean leftmost)，它掺杂着双轴快排和三路快排，使用条件是看这5个点的值是否相等，也就是比如一个数组A，开始时使用了双轴快排，分成了3段A1、A2、A3，可能出现A1后面使用了三路快排（数据里重复元素较多，导致5个点的值全部相等），A2和A3继续使用双轴快排的情况。

复杂度情况

该排序算法的时间复杂度如下：

• 时间复杂度（最好）：O(n)
• 时间复杂度（平均）：O(n * log n)
• 时间复杂度（最差）：O(n * log n)

该排序算法的空间复杂度最差为O(n * log n)。

该排序算法为不稳定排序算法。

动图演示

这个排序算法就不展示动图了。其排序流程图大致如下：

总结

关于该算法的主要内容基本就介绍到这里了，有兴趣的同学可以看看源代码，了解下算法中每个排序（优化排序）具体的实现过程。

可以看到JDK源代码中对于一些常用方法、工具类，是有大量优化的。

源码地址

上述所提到的所有Java代码均可见于我的Github

参考资料

• JDK1.8 java.util.DualPivotQuicksort 源码
• JDK9.0 java.util.DualPivotQuicksort 源码
• JDK11.0 java.util.DualPivotQuicksort 源码