学员优秀博文赏析：双基准快速排序实现

我一般是不会去主动碰算法问题的。上学的时候一个算法复杂度就把我搞烦了，还想让我去搞算法本身？我是这么安慰自己的：反正写 CRUD又不需要算法，这辈子都不需要算法的。好吧，其实我是一直不肯承认，不去动算法的原因只有三个字：因为笨！

所以你去看，一般招算法工程师的都是大厂，而且，薪水永远稳居 程序员 工资的顶端位置。流传在开发界的鄙视链中，算法工程师一直稳居金字塔顶端，俯睨众生。

故，当学员之中出现第一篇涉足算法，还写的有模有样的的博客的时候，我是有点吃惊的。虽然只是基础算法，但这篇博客中除了涉及到了算法本身，还涉及到了源码分析，同时为了保证论证的完整性，文末还给出了 reference。

这几乎是一个优秀程序员都会具备的素质，可是我们的这位学员才学习了 Java一个月啊~~。直到我看到学员本尊的时候，一切都释然了。你们来感受下他的头像，我可以保证确实是本尊无疑：

隐藏的大侠都应该是如此的。

来，现在我们就来欣赏这位同学的博文，师徒班常同学的《Java中双基准快速排序方法(DualPivotQuicksort.sort())的具体实现》：

====================================================================================================

在 Java 语言的Arrays类下提供了一系列排序(sort)方法,帮助使用者对各种不同数据类型的数组进行排序. 在1.7之后的版本中, Arrays.sort()方法在操作过程中实际调用的是DualPivotQuicksort类下的sort方法,DualPivotQuicksort和Arrays一样,都在java.util包下,按字面翻译过来,就是双(Dual)基准(Pivot)快速排序(Quicksort)算法.

双基准快速 排序算法 于2009年由Vladimir Yaroslavskiy提出,是对经典快速排序(Classic Quicksort)进行优化后的一个版本, Java自1.7开始,均使用此方法作为默认排序算法. 接下来,本文就将对此方法的具体实现过程进行简单的介绍.

在正式进入对DualPivotQuicksort的介绍之前,我们先来对经典快速排序的实现思路进行一下简单的了解:

经典快速排序在操作过程中首先会从数组中选取一个基准(Pivot),这个基准可以是数组中的任意一个元素;

随后,将这个数组所有其他元素和Pivot进行比较,比Pivot小的数放在左侧,大的数放在右侧;

如此,我们就在Pivot的左侧和右侧各得到了一个新的数组;

接下来我们再在这两个新的数组中各自选取新的Pivot,把小的放在左侧,大的放在右侧,循环往复,最终就会得到由小到大顺序排列的数组.

下图(via wiki)便是Quicksort的一个具体实现:

在对Quicksort的基本思路有了一定了解之后,下一步我们就来看Java中DualPivotQuicksort.sort是如何实现的;

本方法在实际工作过程中, 并不是对任何传入的数组都直接进行快速排序, 而是会先对数组进行一系列测试, 然后根据数组的具体情况选择最适合的算法来进行排序,下面我们结合源码来看:

首先, 以一个int数组为例,

当一个数组int[] a被传入DualPivotQuicksort.sort()时,该方法还会要求其他一系列参数:

static void sort(int[] a, int left, int right, int[] work, int workBase, int workLen)

其中,int[] a是需被排序的int数组, left和right是该数组中需要被排序的部分的左右界限. 而后面的work, workBase和workLen三个参数其实并不会参与双基准快速排序, 而是当系统认为本数组更适合使用归并排序(merge sort)的时候, 供归并排序使用.

但是,在实际使用中,我们并不希望为了排序设置这么多的参数,因此:

Arrays.sort()在调用DualPivotQuicksort.sort()之前,对int数组的排序提供了两种参数列表:

public static void sort(int[] a)

直接对int[] a 进行排序,以及:

public static void sort(int[] a, int fromIndex, int toIndex)

对int[] a 中从fromIndex到toIndex(包头不包尾)之间的元素进行排序.

在这里,Arrays.sort()会自动将int[] work, int workBase, int workLen设置为null,0,0 省去了使用者的麻烦.

紧接着,DualPivotQuicksort.sort()会对传入的int数组进行检测, 具体流程如下:

这里先贴上整个方法的完整源码, 然后按上图中的流程逐步分析, 只想看DualPivotQuicksort的话可以直接跳到下面第7点:

/**
 * Sorts the specified range of the array using the given
 * workspace array slice if possible for merging
 *
 * @param a the array to be sorted
 * @param left the index of the first element, inclusive, to be sorted
 * @param right the index of the last element, inclusive, to be sorted
 * @param work a workspace array (slice)
 * @param workBase origin of usable space in work array
 * @param workLen usable size of work array
 */
static void sort(int[] a, int left, int right,
                 int[] work, int workBase, int workLen) {
    // Use Quicksort on small arrays
    if (right - left < QUICKSORT_THRESHOLD) {
        sort(a, left, right, true);
        return;
    }

    /*
     * Index run[i] is the start of i-th run
     * (ascending or descending sequence).
     */
    int[] run = new int[MAX_RUN_COUNT + 1];
    int count = 0; run[0] = left;

    // Check if the array is nearly sorted
    for (int k = left; k < right; run[count] = k) {
        if (a[k] < a[k + 1]) { // ascending
            while (++k <= right && a[k - 1] <= a[k]);
        } else if (a[k] > a[k + 1]) { // descending
            while (++k <= right && a[k - 1] >= a[k]);
            for (int lo = run[count] - 1, hi = k; ++lo < --hi; ) {
                int t = a[lo]; a[lo] = a[hi]; a[hi] = t;
            }
        } else { // equal
            for (int m = MAX_RUN_LENGTH; ++k <= right && a[k - 1] == a[k]; ) {
                if (--m == 0) {
                    sort(a, left, right, true);
                    return;
                }
            }
        }

        /*
         * The array is not highly structured,
         * use Quicksort instead of merge sort.
         */
        if (++count == MAX_RUN_COUNT) {
            sort(a, left, right, true);
            return;
        }
    }

    // Check special cases
    // Implementation note: variable "right" is increased by 1.
    if (run[count] == right++) { // The last run contains one element
        run[++count] = right;
    } else if (count == 1) { // The array is already sorted
        return;
    }

    // Determine alternation base for merge
    byte odd = 0;
    for (int n = 1; (n <<= 1) < count; odd ^= 1);

    // Use or create temporary array b for merging
    int[] b;                 // temp array; alternates with a
    int ao, bo;              // array offsets from 'left'
    int blen = right - left; // space needed for b
    if (work == null || workLen < blen || workBase + blen > work.length) {
        work = new int[blen];
        workBase = 0;
    }
    if (odd == 0) {
        System.arraycopy(a, left, work, workBase, blen);
        b = a;
        bo = 0;
        a = work;
        ao = workBase - left;
    } else {
        b = work;
        ao = 0;
        bo = workBase - left;
    }

    // Merging
    for (int last; count > 1; count = last) {
        for (int k = (last = 0) + 2; k <= count; k += 2) {
            int hi = run[k], mi = run[k - 1];
            for (int i = run[k - 2], p = i, q = mi; i < hi; ++i) {
                if (q >= hi || p < mi && a[p + ao] <= a[q + ao]) {
                    b[i + bo] = a[p++ + ao];
                } else {
                    b[i + bo] = a[q++ + ao];
                }
            }
            run[++last] = hi;
        }
        if ((count & 1) != 0) {
            for (int i = right, lo = run[count - 1]; --i >= lo;
                b[i + bo] = a[i + ao]
            );
            run[++last] = right;
        }
        int[] t = a; a = b; b = t;
        int o = ao; ao = bo; bo = o;
    }
}

/**
 * Sorts the specified range of the array by Dual-Pivot Quicksort.
 *
 * @param a the array to be sorted
 * @param left the index of the first element, inclusive, to be sorted
 * @param right the index of the last element, inclusive, to be sorted
 * @param leftmost indicates if this part is the leftmost in the range
 */
private static void sort(int[] a, int left, int right, boolean leftmost) {
    int length = right - left + 1;

    // Use insertion sort on tiny arrays
    if (length < INSERTION_SORT_THRESHOLD) {
        if (leftmost) {
            /*
             * Traditional (without sentinel) insertion sort,
             * optimized for server VM, is used in case of
             * the leftmost part.
             */
            for (int i = left, j = i; i < right; j = ++i) {
                int ai = a[i + 1];
                while (ai < a[j]) {
                    a[j + 1] = a[j];
                    if (j-- == left) {
                        break;
                    }
                }
                a[j + 1] = ai;
            }
        } else {
            /*
             * Skip the longest ascending sequence.
             */
            do {
                if (left >= right) {
                    return;
                }
            } while (a[++left] >= a[left - 1]);

            /*
             * Every element from adjoining part plays the role
             * of sentinel, therefore this allows us to avoid the
             * left range check on each iteration. Moreover, we use
             * the more optimized algorithm, so called pair insertion
             * sort, which is faster (in the context of Quicksort)
             * than traditional implementation of insertion sort.
             */
            for (int k = left; ++left <= right; k = ++left) {
                int a1 = a[k], a2 = a[left];

                if (a1 < a2) {
                    a2 = a1; a1 = a[left];
                }
                while (a1 < a[--k]) {
                    a[k + 2] = a[k];
                }
                a[++k + 1] = a1;

                while (a2 < a[--k]) {
                    a[k + 1] = a[k];
                }
                a[k + 1] = a2;
            }
            int last = a[right];

            while (last < a[--right]) {
                a[right + 1] = a[right];
            }
            a[right + 1] = last;
        }
        return;
    }

    // Inexpensive approximation of length / 7
    int seventh = (length >> 3) + (length >> 6) + 1;

    /*
     * Sort five evenly spaced elements around (and including) the
     * center element in the range. These elements will be used for
     * pivot selection as described below. The choice for spacing
     * these elements was empirically determined to work well on
     * a wide variety of inputs.
     */
    int e3 = (left + right) >>> 1; // The midpoint
    int e2 = e3 - seventh;
    int e1 = e2 - seventh;
    int e4 = e3 + seventh;
    int e5 = e4 + seventh;

    // Sort these elements using insertion sort
    if (a[e2] < a[e1]) { int t = a[e2]; a[e2] = a[e1]; a[e1] = t; }

    if (a[e3] < a[e2]) { int t = a[e3]; a[e3] = a[e2]; a[e2] = t;
        if (t < a[e1]) { a[e2] = a[e1]; a[e1] = t; }
    }
    if (a[e4] < a[e3]) { int t = a[e4]; a[e4] = a[e3]; a[e3] = t;
        if (t < a[e2]) { a[e3] = a[e2]; a[e2] = t;
            if (t < a[e1]) { a[e2] = a[e1]; a[e1] = t; }
        }
    }
    if (a[e5] < a[e4]) { int t = a[e5]; a[e5] = a[e4]; a[e4] = t;
        if (t < a[e3]) { a[e4] = a[e3]; a[e3] = t;
            if (t < a[e2]) { a[e3] = a[e2]; a[e2] = t;
                if (t < a[e1]) { a[e2] = a[e1]; a[e1] = t; }
            }
        }
    }

    // Pointers
    int less  = left;  // The index of the first element of center part
    int great = right; // The index before the first element of right part

    if (a[e1] != a[e2] && a[e2] != a[e3] && a[e3] != a[e4] && a[e4] != a[e5]) {
        /*
         * Use the second and fourth of the five sorted elements as pivots.
         * These values are inexpensive approximations of the first and
         * second terciles of the array. Note that pivot1 <= pivot2.
         */
        int pivot1 = a[e2];
        int pivot2 = a[e4];

        /*
         * The first and the last elements to be sorted are moved to the
         * locations formerly occupied by the pivots. When partitioning
         * is complete, the pivots are swapped back into their final
         * positions, and excluded from subsequent sorting.
         */
        a[e2] = a[left];
        a[e4] = a[right];

        /*
         * Skip elements, which are less or greater than pivot values.
         */
        while (a[++less] < pivot1);
        while (a[--great] > pivot2);

        /*
         * Partitioning:
         *
         *   left part           center part                   right part
         * +--------------------------------------------------------------+
         * |  < pivot1  |  pivot1 <= && <= pivot2  |    ?    |  > pivot2  |
         * +--------------------------------------------------------------+
         *               ^                          ^       ^
         *               |                          |       |
         *              less                        k     great
         *
         * Invariants:
         *
         *              all in (left, less)   < pivot1
         *    pivot1 <= all in [less, k)     <= pivot2
         *              all in (great, right) > pivot2
         *
         * Pointer k is the first index of ?-part.
         */
        outer:
        for (int k = less - 1; ++k <= great; ) {
            int ak = a[k];
            if (ak < pivot1) { // Move a[k] to left part
                a[k] = a[less];
                /*
                 * Here and below we use "a[i] = b; i++;" instead
                 * of "a[i++] = b;" due to performance issue.
                 */
                a[less] = ak;
                ++less;
            } else if (ak > pivot2) { // Move a[k] to right part
                while (a[great] > pivot2) {
                    if (great-- == k) {
                        break outer;
                    }
                }
                if (a[great] < pivot1) { // a[great] <= pivot2
                    a[k] = a[less];
                    a[less] = a[great];
                    ++less;
                } else { // pivot1 <= a[great] <= pivot2
                    a[k] = a[great];
                }
                /*
                 * Here and below we use "a[i] = b; i--;" instead
                 * of "a[i--] = b;" due to performance issue.
                 */
                a[great] = ak;
                --great;
            }
        }

        // Swap pivots into their final positions
        a[left]  = a[less  - 1]; a[less  - 1] = pivot1;
        a[right] = a[great + 1]; a[great + 1] = pivot2;

        // Sort left and right parts recursively, excluding known pivots
        sort(a, left, less - 2, leftmost);
        sort(a, great + 2, right, false);

        /*
         * If center part is too large (comprises > 4/7 of the array),
         * swap internal pivot values to ends.
         */
        if (less < e1 && e5 < great) {
            /*
             * Skip elements, which are equal to pivot values.
             */
            while (a[less] == pivot1) {
                ++less;
            }

            while (a[great] == pivot2) {
                --great;
            }

            /*
             * Partitioning:
             *
             *   left part         center part                  right part
             * +----------------------------------------------------------+
             * | == pivot1 |  pivot1 < && < pivot2  |    ?    | == pivot2 |
             * +----------------------------------------------------------+
             *              ^                        ^       ^
             *              |                        |       |
             *             less                      k     great
             *
             * Invariants:
             *
             *              all in (*,  less) == pivot1
             *     pivot1 < all in [less,  k)  < pivot2
             *              all in (great, *) == pivot2
             *
             * Pointer k is the first index of ?-part.
             */
            outer:
            for (int k = less - 1; ++k <= great; ) {
                int ak = a[k];
                if (ak == pivot1) { // Move a[k] to left part
                    a[k] = a[less];
                    a[less] = ak;
                    ++less;
                } else if (ak == pivot2) { // Move a[k] to right part
                    while (a[great] == pivot2) {
                        if (great-- == k) {
                            break outer;
                        }
                    }
                    if (a[great] == pivot1) { // a[great] < pivot2
                        a[k] = a[less];
                        /*
                         * Even though a[great] equals to pivot1, the
                         * assignment a[less] = pivot1 may be incorrect,
                         * if a[great] and pivot1 are floating-point zeros
                         * of different signs. Therefore in float and
                         * double sorting methods we have to use more
                         * accurate assignment a[less] = a[great].
                         */
                        a[less] = pivot1;
                        ++less;
                    } else { // pivot1 < a[great] < pivot2
                        a[k] = a[great];
                    }
                    a[great] = ak;
                    --great;
                }
            }
        }

        // Sort center part recursively
        sort(a, less, great, false);

    } else { // Partitioning with one pivot
        /*
         * Use the third of the five sorted elements as pivot.
         * This value is inexpensive approximation of the median.
         */
        int pivot = a[e3];

        /*
         * Partitioning degenerates to the traditional 3-way
         * (or "Dutch National Flag") schema:
         *
         *   left part    center part              right part
         * +-------------------------------------------------+
         * |  < pivot  |   == pivot   |     ?    |  > pivot  |
         * +-------------------------------------------------+
         *              ^              ^        ^
         *              |              |        |
         *             less            k      great
         *
         * Invariants:
         *
         *   all in (left, less)   < pivot
         *   all in [less, k)     == pivot
         *   all in (great, right) > pivot
         *
         * Pointer k is the first index of ?-part.
         */
        for (int k = less; k <= great; ++k) {
            if (a[k] == pivot) {
                continue;
            }
            int ak = a[k];
            if (ak < pivot) { // Move a[k] to left part
                a[k] = a[less];
                a[less] = ak;
                ++less;
            } else { // a[k] > pivot - Move a[k] to right part
                while (a[great] > pivot) {
                    --great;
                }
                if (a[great] < pivot) { // a[great] <= pivot
                    a[k] = a[less];
                    a[less] = a[great];
                    ++less;
                } else { // a[great] == pivot
                    /*
                     * Even though a[great] equals to pivot, the
                     * assignment a[k] = pivot may be incorrect,
                     * if a[great] and pivot are floating-point
                     * zeros of different signs. Therefore in float
                     * and double sorting methods we have to use
                     * more accurate assignment a[k] = a[great].
                     */
                    a[k] = pivot;
                }
                a[great] = ak;
                --great;
            }
        }

        /*
         * Sort left and right parts recursively.
         * All elements from center part are equal
         * and, therefore, already sorted.
         */
        sort(a, left, less - 1, leftmost);
        sort(a, great + 1, right, false);
    }
}

DualPivotQuickSort.sort()

1. 判断数组int[] a的长度是否大于常量QUICKSORT_THRESHOLD, 即286:

286是java设定的一个阈值,当数组长度小于此值时, 系统将不再考虑merge sort, 直接将参数传入本类中的另一个私有sort方法进行排序

private static void sort(long[] a, int left, int right, boolean leftmost)

 // Use Quicksort on small arrays
 if (right - left < QUICKSORT_THRESHOLD) {
     sort(a, left, right, true);
     return;
｝

2. 继续判断int[] a的长度是否大于常量INSERTION_SORT_THRESHOLD, 即47:

3. 若数组长度小于47, 则使用insertion sort:

数组传入本类私有的sort方法后, 会继续判断数组长度是否大于47, 若小于此值, 则直接使用insertion sort并返回结果, 因为插入算法并非本文重点, 此处不再展开叙述

int length = right - left + 1;

// Use insertion sort on tiny arrays
if (length < INSERTION_SORT_THRESHOLD) {
    if (leftmost) {
        /*
         * Traditional (without sentinel) insertion sort,
         * optimized for server VM, is used in case of
         * the leftmost part.
         */
        for (int i = left, j = i; i < right; j = ++i) {
            int ai = a[i + 1];
            while (ai < a[j]) {
                a[j + 1] = a[j];
                if (j-- == left) {
                    break;
                }
            }
            a[j + 1] = ai;
        }
    } else {
        /*
         * Skip the longest ascending sequence.
         */
        do {
            if (left >= right) {
                return;
            }
        } while (a[++left] >= a[left - 1]);

        /*
         * Every element from adjoining part plays the role
         * of sentinel, therefore this allows us to avoid the
         * left range check on each iteration. Moreover, we use
         * the more optimized algorithm, so called pair insertion
         * sort, which is faster (in the context of Quicksort)
         * than traditional implementation of insertion sort.
         */
        for (int k = left; ++left <= right; k = ++left) {
            int a1 = a[k], a2 = a[left];

            if (a1 < a2) {
                a2 = a1; a1 = a[left];
            }
            while (a1 < a[--k]) {
                a[k + 2] = a[k];
            }
            a[++k + 1] = a1;

            while (a2 < a[--k]) {
                a[k + 1] = a[k];
            }
            a[k + 1] = a2;
        }
        int last = a[right];

        while (last < a[--right]) {
            a[right + 1] = a[right];
        }
        a[right + 1] = last;
    }
    return;
}

判断数组长度是否小于47,若小于则直接使用插入排序

值得注意的是, java在这里提供了两种不同的插入排序算法, 当传入的参数leftmost真假值不同时, 会使用不同的算法.

leftmost代表的是本次传入的数组是否是从最初的int[] a的最左侧left开始的, 因为本方法在整个排序过程中可能会针对数组的不同部分被多次调用, 因此leftmost有可能为false.

Quicksort的情况我们放到最后再谈, 这里先回过来看第一步判断中数组长度大于286的情形, 这种情况下, 系统会

4.继续判断该数组是否已经高度结构化(即已经接近排序完成):

这里的基本思路是这样的:

a. 定义一个常量MAX_RUN_COUNT = 67;

b. 定义一个计数器int count = 0; 定义一个数组int[] run 使之长度为MAX_RUN_COUNT + 1;

c. 令run[0] = left, 然后从传入数组的最左侧left开始遍历, 若数组的前n个元素均为升序/降序排列, 而第n + 1个元素的升/降序发生了改变, 则将第n个元素的索引存入run[1], 同时++count, 此时count的值为1;

d. 从n + 1开始继续遍历, 直至升/降序再次改变, 再将此处的索引存入run[2], ++count, 此时count的值为2, 以此类推...

......

e. 若将整个数组全部遍历完成后, count仍然小于MAX_RUN_COUNT (即整个数组升降序改变的次数低于67次), 证明该数组是高度结构化的, 则使用merge sort进行排序;

若count == MAX_RUN_COUNT时, 还未完成对数组的遍历, 则证明数组并非高度结构化, 则调用前文所述私有sort方法进行quicksort.

/*
 * Index run[i] is the start of i-th run
 * (ascending or descending sequence).
 */
int[] run = new int[MAX_RUN_COUNT + 1];
int count = 0; run[0] = left;

// Check if the array is nearly sorted
for (int k = left; k < right; run[count] = k) {
    if (a[k] < a[k + 1]) { // ascending
        while (++k <= right && a[k - 1] <= a[k]);
    } else if (a[k] > a[k + 1]) { // descending
        while (++k <= right && a[k - 1] >= a[k]);
        for (int lo = run[count] - 1, hi = k; ++lo < --hi; ) {
            int t = a[lo]; a[lo] = a[hi]; a[hi] = t;
        }
    } else { // equal
        for (int m = MAX_RUN_LENGTH; ++k <= right && a[k - 1] == a[k]; ) {
            if (--m == 0) {
                sort(a, left, right, true);
                return;
            }
        }
    }

    /*
     * The array is not highly structured,
     * use Quicksort instead of merge sort.
     */
    if (++count == MAX_RUN_COUNT) {
        sort(a, left, right, true);
        return;
    }
}

判断该数组是否已经高度结构化

5. 判断该数组是否是是已经排列好的:

若该数组是高度结构化的, 在使用merge sort进行排序之前, 会先检验数组是否本身就是排序好的, 思路很简单, 如果在前面的检测中一次就完成了遍历, 就证明该数组是排序好的, 则直接返回结果:

// Check special cases
// Implementation note: variable "right" is increased by 1.
if (run[count] == right++) { // The last run contains one element
    run[++count] = right;
} else if (count == 1) { // The array is already sorted
    return;
}

判断数组是否本来就是排列好的

*当然, 在具体实现中还有不少其他要考虑的因素, 有兴趣了解的话可以结合上一部分代码进行阅读.

6. 进行归并排序(merge sort):

此处不再展开叙述, 值得注意的是, 由于归并算法在操作过程中需要使用一块额外的存储空间, 本方法参数列表中要求的work, workBase和workLen三个参数就是在此处使用的:

// Determine alternation base for merge
byte odd = 0;
for (int n = 1; (n <<= 1) < count; odd ^= 1);

// Use or create temporary array b for merging
int[] b;                 // temp array; alternates with a
int ao, bo;              // array offsets from 'left'
int blen = right - left; // space needed for b
if (work == null || workLen < blen || workBase + blen > work.length) {
    work = new int[blen];
    workBase = 0;
}
if (odd == 0) {
    System.arraycopy(a, left, work, workBase, blen);
    b = a;
    bo = 0;
    a = work;
    ao = workBase - left;
} else {
    b = work;
    ao = 0;
    bo = workBase - left;
}

// Merging
for (int last; count > 1; count = last) {
    for (int k = (last = 0) + 2; k <= count; k += 2) {
        int hi = run[k], mi = run[k - 1];
        for (int i = run[k - 2], p = i, q = mi; i < hi; ++i) {
            if (q >= hi || p < mi && a[p + ao] <= a[q + ao]) {
                b[i + bo] = a[p++ + ao];
            } else {
                b[i + bo] = a[q++ + ao];
            }
        }
        run[++last] = hi;
    }
    if ((count & 1) != 0) {
        for (int i = right, lo = run[count - 1]; --i >= lo;
            b[i + bo] = a[i + ao]
        );
        run[++last] = right;
    }
    int[] t = a; a = b; b = t;
    int o = ao; ao = bo; bo = o;
}

merge sort

7. 进行双基准快速排序(dual pivot quicksort):

只有在上述情况都不满足的情况下, 本方法才会使用双基准快速排序算法进行排序,

算法本身的思路并不复杂, 和经典快速排序相差不大, 顾名思义, 比起经典快排, 该算法选取了两个Pivot, 我们姑且称之为P1和P2.

P1和P2都从数组中选出, P1在P2的右侧, 且P1必须小于P2, 如果不是, 则交换P1和P2的值;

接下来令数组中的每一个元素和基准进行比较, 比P1小的放在P1左边, 比P2大的放在P2右边, 介于两者之间的放在中间.

这样, 最终我们会的得到三个数组, 比P1小元素构成的数组, 介于P1和P2之间的元素构成的数组, 以及比P2大的元素构成的数组.

最后, 递归地对这三个数组进行排序, 最终得到排序完成的结果.

思路上虽然, 并不复杂, 但Java为了尽可能的提高效率, 在对这个算法进行实现的过程中增加了非常多的细节, 下面我们就来大致看一下其中的部分内容:

// Inexpensive approximation of length / 7
int seventh = (length >> 3) + (length >> 6) + 1;

/*
 * Sort five evenly spaced elements around (and including) the
 * center element in the range. These elements will be used for
 * pivot selection as described below. The choice for spacing
 * these elements was empirically determined to work well on
 * a wide variety of inputs.
 */
int e3 = (left + right) >>> 1; // The midpoint
int e2 = e3 - seventh;
int e1 = e2 - seventh;
int e4 = e3 + seventh;
int e5 = e4 + seventh;

// Sort these elements using insertion sort
if (a[e2] < a[e1]) { int t = a[e2]; a[e2] = a[e1]; a[e1] = t; }

if (a[e3] < a[e2]) { int t = a[e3]; a[e3] = a[e2]; a[e2] = t;
    if (t < a[e1]) { a[e2] = a[e1]; a[e1] = t; }
}
if (a[e4] < a[e3]) { int t = a[e4]; a[e4] = a[e3]; a[e3] = t;
    if (t < a[e2]) { a[e3] = a[e2]; a[e2] = t;
        if (t < a[e1]) { a[e2] = a[e1]; a[e1] = t; }
    }
}
if (a[e5] < a[e4]) { int t = a[e5]; a[e5] = a[e4]; a[e4] = t;
    if (t < a[e3]) { a[e4] = a[e3]; a[e3] = t;
        if (t < a[e2]) { a[e3] = a[e2]; a[e2] = t;
            if (t < a[e1]) { a[e2] = a[e1]; a[e1] = t; }
        }
    }
}

// Pointers
int less  = left;  // The index of the first element of center part
int great = right; // The index before the first element of right part

if (a[e1] != a[e2] && a[e2] != a[e3] && a[e3] != a[e4] && a[e4] != a[e5]) {
    /*
     * Use the second and fourth of the five sorted elements as pivots.
     * These values are inexpensive approximations of the first and
     * second terciles of the array. Note that pivot1 <= pivot2.
     */
    int pivot1 = a[e2];
    int pivot2 = a[e4];

    /*
     * The first and the last elements to be sorted are moved to the
     * locations formerly occupied by the pivots. When partitioning
     * is complete, the pivots are swapped back into their final
     * positions, and excluded from subsequent sorting.
     */
    a[e2] = a[left];
    a[e4] = a[right];

    /*
     * Skip elements, which are less or greater than pivot values.
     */
    while (a[++less] < pivot1);
    while (a[--great] > pivot2);

    /*
     * Partitioning:
     *
     *   left part           center part                   right part
     * +--------------------------------------------------------------+
     * |  < pivot1  |  pivot1 <= && <= pivot2  |    ?    |  > pivot2  |
     * +--------------------------------------------------------------+
     *               ^                          ^       ^
     *               |                          |       |
     *              less                        k     great
     *
     * Invariants:
     *
     *              all in (left, less)   < pivot1
     *    pivot1 <= all in [less, k)     <= pivot2
     *              all in (great, right) > pivot2
     *
     * Pointer k is the first index of ?-part.
     */
    outer:
    for (int k = less - 1; ++k <= great; ) {
        int ak = a[k];
        if (ak < pivot1) { // Move a[k] to left part
            a[k] = a[less];
            /*
             * Here and below we use "a[i] = b; i++;" instead
             * of "a[i++] = b;" due to performance issue.
             */
            a[less] = ak;
            ++less;
        } else if (ak > pivot2) { // Move a[k] to right part
            while (a[great] > pivot2) {
                if (great-- == k) {
                    break outer;
                }
            }
            if (a[great] < pivot1) { // a[great] <= pivot2
                a[k] = a[less];
                a[less] = a[great];
                ++less;
            } else { // pivot1 <= a[great] <= pivot2
                a[k] = a[great];
            }
            /*
             * Here and below we use "a[i] = b; i--;" instead
             * of "a[i--] = b;" due to performance issue.
             */
            a[great] = ak;
            --great;
        }
    }

    // Swap pivots into their final positions
    a[left]  = a[less  - 1]; a[less  - 1] = pivot1;
    a[right] = a[great + 1]; a[great + 1] = pivot2;

    // Sort left and right parts recursively, excluding known pivots
    sort(a, left, less - 2, leftmost);
    sort(a, great + 2, right, false);

    /*
     * If center part is too large (comprises > 4/7 of the array),
     * swap internal pivot values to ends.
     */
    if (less < e1 && e5 < great) {
        /*
         * Skip elements, which are equal to pivot values.
         */
        while (a[less] == pivot1) {
            ++less;
        }

        while (a[great] == pivot2) {
            --great;
        }

        /*
         * Partitioning:
         *
         *   left part         center part                  right part
         * +----------------------------------------------------------+
         * | == pivot1 |  pivot1 < && < pivot2  |    ?    | == pivot2 |
         * +----------------------------------------------------------+
         *              ^                        ^       ^
         *              |                        |       |
         *             less                      k     great
         *
         * Invariants:
         *
         *              all in (*,  less) == pivot1
         *     pivot1 < all in [less,  k)  < pivot2
         *              all in (great, *) == pivot2
         *
         * Pointer k is the first index of ?-part.
         */
        outer:
        for (int k = less - 1; ++k <= great; ) {
            int ak = a[k];
            if (ak == pivot1) { // Move a[k] to left part
                a[k] = a[less];
                a[less] = ak;
                ++less;
            } else if (ak == pivot2) { // Move a[k] to right part
                while (a[great] == pivot2) {
                    if (great-- == k) {
                        break outer;
                    }
                }
                if (a[great] == pivot1) { // a[great] < pivot2
                    a[k] = a[less];
                    /*
                     * Even though a[great] equals to pivot1, the
                     * assignment a[less] = pivot1 may be incorrect,
                     * if a[great] and pivot1 are floating-point zeros
                     * of different signs. Therefore in float and
                     * double sorting methods we have to use more
                     * accurate assignment a[less] = a[great].
                     */
                    a[less] = pivot1;
                    ++less;
                } else { // pivot1 < a[great] < pivot2
                    a[k] = a[great];
                }
                a[great] = ak;
                --great;
            }
        }
    }

    // Sort center part recursively
    sort(a, less, great, false);

} else { // Partitioning with one pivot
    /*
     * Use the third of the five sorted elements as pivot.
     * This value is inexpensive approximation of the median.
     */
    int pivot = a[e3];

    /*
     * Partitioning degenerates to the traditional 3-way
     * (or "Dutch National Flag") schema:
     *
     *   left part    center part              right part
     * +-------------------------------------------------+
     * |  < pivot  |   == pivot   |     ?    |  > pivot  |
     * +-------------------------------------------------+
     *              ^              ^        ^
     *              |              |        |
     *             less            k      great
     *
     * Invariants:
     *
     *   all in (left, less)   < pivot
     *   all in [less, k)     == pivot
     *   all in (great, right) > pivot
     *
     * Pointer k is the first index of ?-part.
     */
    for (int k = less; k <= great; ++k) {
        if (a[k] == pivot) {
            continue;
        }
        int ak = a[k];
        if (ak < pivot) { // Move a[k] to left part
            a[k] = a[less];
            a[less] = ak;
            ++less;
        } else { // a[k] > pivot - Move a[k] to right part
            while (a[great] > pivot) {
                --great;
            }
            if (a[great] < pivot) { // a[great] <= pivot
                a[k] = a[less];
                a[less] = a[great];
                ++less;
            } else { // a[great] == pivot
                /*
                 * Even though a[great] equals to pivot, the
                 * assignment a[k] = pivot may be incorrect,
                 * if a[great] and pivot are floating-point
                 * zeros of different signs. Therefore in float
                 * and double sorting methods we have to use
                 * more accurate assignment a[k] = a[great].
                 */
                a[k] = pivot;
            }
            a[great] = ak;
            --great;
        }
    }

    /*
     * Sort left and right parts recursively.
     * All elements from center part are equal
     * and, therefore, already sorted.
     */
    sort(a, left, less - 1, leftmost);
    sort(a, great + 1, right, false);
}

先贴上这部分完整的代码

首先, 也是本方法中最具特色的部分, 就是对Pivot的选取:

在这里, 系统会先通过位运算获取数组长度的1/7的近似值(位运算无法精确表示1/7)

int seventh = (length >> 3) + (length >> 6) + 1;

然后获取本数组中间位置的索引e3:

int e3 = (left + right) >>> 1; // The midpoint

在中间位置的左右1/7, 2/7处各获取两个索引(e1, e2, e4, e5):

int e2 = e3 - seventh;
int e1 = e2 - seventh;
int e4 = e3 + seventh;
int e5 = e4 + seventh;

将这五个索引对应的值用插入算法进行有小到大的排序后, 再放回五个索引的位置

// Sort these elements using insertion sort
if (a[e2] < a[e1]) { int t = a[e2]; a[e2] = a[e1]; a[e1] = t; }

if (a[e3] < a[e2]) { int t = a[e3]; a[e3] = a[e2]; a[e2] = t;
    if (t < a[e1]) { a[e2] = a[e1]; a[e1] = t; }
}
if (a[e4] < a[e3]) { int t = a[e4]; a[e4] = a[e3]; a[e3] = t;
    if (t < a[e2]) { a[e3] = a[e2]; a[e2] = t;
        if (t < a[e1]) { a[e2] = a[e1]; a[e1] = t; }
    }
}
if (a[e5] < a[e4]) { int t = a[e5]; a[e5] = a[e4]; a[e4] = t;
    if (t < a[e3]) { a[e4] = a[e3]; a[e3] = t;
        if (t < a[e2]) { a[e3] = a[e2]; a[e2] = t;
            if (t < a[e1]) { a[e2] = a[e1]; a[e1] = t; }
        }
    }
}

对索引的值排序

接下来进行判断, 若这五个索引对应的元素值各不相同, 则选取e2的值作为Pivot1, e4的值作为Pivot2(特别注意基准是值而不是元素), 然后进行双基准快排, Java声称这种选法比其他方式更加高效;

但如果这五个值中有相同的存在, 则本轮排序选取e3的值作为Pivot, 进行单基准快排, 同样, Java声称这种选取方式要比随机选取更加高效.

至于具体排序的部分, 其实并没有太多可以叙述的内容, 值得注意的是, 由于在递归的过程中会不断地调用私有的sort方法, 所以在递归中获得的子数组长度小47时, 会改为调用插入排序.

此外, 源码中还有一些可以借鉴的用来提高效率的小窍门, 比如在对元素进行交换位置之前先过滤掉头尾处已经在正确位置的元素等, 有兴趣的话可以再对源码进行仔细阅读.

至此, 关于DualPivotQuicksort.sort()的大致实现过程的介绍也就基本结束了, 最后我们不妨进行一个简单的测试, 看看它是否真的比经典快排更加高效:

首先, 我们先来测试一下经典Quicksort的性能, 对包含一亿个Random.nextInt()随机数的数组进行排序, 重复十次求平均排序运行时间,

具体代码如下, 其中经典快排的代码摘自wiki快速排序条目:

import java.util.Date;
import java.util.Random;

public class Test {
    public static void main(String[] args) {
        //定义time记录总耗时
        int time = 0;
        
        //测试十次
        for (int i = 0; i < 10; i++) {
            //生成一个包含一亿个Random.nextInt()随机数的数组
            Random r = new Random();
            int[] arr = new int[100000000];
            for (int j : arr) {
                arr[j] = r.nextInt();
            }
            
            //使用经典快排对数组进行排序,并获取运行时间
            long t1 = new Date().getTime();
            qSort(arr, 0, arr.length - 1);
            long t2 = new Date().getTime();
            
            //输出运行时间
            System.out.print((int)(t2 - t1) + " ");
            //对每次运行时间求和
            time += (int)(t2 - t1);
             
        }
        //输出十次排序平均运行时间
        System.out.println(time / 10);
    }
    
    public static void qSort(int[] arr, int head, int tail) {
        if (head >= tail || arr == null || arr.length <= 1) {
            return;
        }
        int i = head, j = tail, pivot = arr[(head + tail) / 2];
        while (i <= j) {
            while (arr[i] < pivot) {
                ++i;
            }
            while (arr[j] > pivot) {
                --j;
            }
            if (i < j) {
                int t = arr[i];
                arr[i] = arr[j];
                arr[j] = t;
                ++i;
                --j;
            } else if (i == j) {
                ++i;
            }
        }
        qSort(arr, head, j);
        qSort(arr, i, tail);
    }
}

经典快排测试

经过测试, 我们得到结果, 十次排序的运行时间分别为2405 3017 3021 2972 2993 2967 2992 2954 3001 2984(ms),

平均时间则是2930ms

接下来, 再对DualPivotQuicksort.sort()使用同样的方法进行测试, 直接用Arrays.sort()进行排序,代码如下:

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

public class Test2 {
    public static void main(String[] args) {
        //定义time记录总耗时
        int time = 0;
        
        //测试十次
        for (int i = 0; i < 10; i++) {
            //生成一个包含一亿个Random.nextInt()随机数的数组
            Random r = new Random();
            int[] arr = new int[100000000];
            for (int j : arr) {
                arr[j] = r.nextInt();
            }
            
            //使用经典快排对数组进行排序,并获取运行时间
            long t1 = new Date().getTime();
            Arrays.sort(arr);
            long t2 = new Date().getTime();
            
            //输出运行时间
            System.out.print((int)(t2 - t1) + " ");
            //对每次运行时间求和
            time += (int)(t2 - t1);
             
        }
        //输出十次排序平均运行时间
        System.out.println(time / 10);
    }
}

双基准快排测试

十次运行时间分别是115 16 31 21 28 118 62 78 78 63(ms).

平均时间则达到了惊人的61ms.

当然, 之所以会产生如此巨大的差距, 一方面是因为双基准快排本身性能更加优秀, 另一方也是因为Java对该方法进行了大量的优化, 而选取测试的经典快排则相当粗糙, 只是这种算法思想的体现, 并不是说两种算法本身性能上存在着这种程度的差距.

Reference:

Quicksort - wiki

https://en.wikipedia.org/wiki/Quicksort

快速排序 - wiki

https://zh.wikipedia.org/zh-cn/%E5%BF%AB%E9%80%9F%E6%8E%92%E5%BA%8F

java.util.Arrays

java.util.DualPivotQuicksort