您的位置:首页 > 编程语言 > Java开发

JDK DualPivotQuicksort 源码解析

2016-11-23 17:45 459 查看
参考自:论文,Dual-Pivot Quicksort algorithm ,by Vladimir Yaroslavskiy。
http://www.sytarena.com/javajswz/20140217/1329.html 
DualPivotQuicksort是JDK1.7开始的采用的快速排序算法。

一般的快速排序采用一个枢轴来把一个数组划分成两半,然后递归之。

大量经验数据表面,采用两个枢轴来划分成3份的算法更高效,这就是DualPivotQuicksort。


算法思想

选出两个枢轴P1和P2,需要3个指针L,K,G。

3个指针的作用如下图:



算法为以下的步骤:

1、小于27的数组,使用插入排序(或47)。

2、选择枢轴P1和P2。(假设使用数组头和尾)。

3、P1需要小于P2,否者交换。

现在数组被分成4份,left到L的小于P1的数,L到K的大于P1小于P2的数,G到rigth的大于P2的数,待处理的K到G的中间的数(逐步被处理到前3个区域中)。

4、L从开始初始化直至不小于P1,K初始化为L-1,G从结尾初始化直至不大于P2。K是主移动的指针,来一步一步吞噬中间区域。

****当大于P1小于P2,K++。

****当小于P1,交换L和K的数,L++,K++。

****当大于P2,如果G的数小于P1,把L上的数放在K上,把G的数放在L上,L++,再把K以前的数放在G上,G--,K++,完成一次L,K,G的互相交换。否则交换K和G,并G--,K++。

5、递归4。

6、交换P1到L-1上。交换P2到G+1上。

7、递归之。


JDK源码 

流程图:




TimSort

直接调用的sort第一级:

public static void sort(int[] a, int left, int right) {
// Use Quicksort on small arrays
if (right - left < QUICKSORT_THRESHOLD) {//门限为286
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
if (run[count] == right++) { // The last run contains one element
run[++count] = right;
} else if (count == 1) { // The array is already sorted
return;
}

/*
* Create temporary array, which is used for merging.
* Implementation note: variable "right" is increased by 1.
*/
int[] b; byte odd = 0;
for (int n = 1; (n <<= 1) < count; odd ^= 1);

if (odd == 0) {
b = a; a = new int[b.length];
for (int i = left - 1; ++i < right; a[i] = b[i]);
} else {
b = new int[a.length];
}

// 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] <= a[q]) {
b[i] = a[p++];
} else {
b[i] = a[q++];
}
}
run[++last] = hi;
}
if ((count & 1) != 0) {
for (int i = right, lo = run[count - 1]; --i >= lo;
b[i] = a[i]
);
run[++last] = right;
}
int[] t = a; a = b; b = t;
}
}

1、当小于286时,直接使用双枢轴快速排序。

2、大于286时,先使用TimSort的思想,找出正序或者倒序的数(run的栈),然后合并各个run,最后完成TimSort。 

3、在找出run的时候,找出的run个数大于67时,即认为它是一个比较乱序的数组,就直接采用双枢轴快速排序。


双元素插入排序

下面看看双枢轴快速排序的实现(代码太长,分解之):

/**
* 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) {//47个
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个时,使用插入排序。

参数a为需要排序的数组,left代表需要排序的数组区间中最左边元素的索引,right代表区间中最右边元素的索引,leftmost代表该区间是否是数组中最左边的区间。举个例子:

数组:[2, 4, 8, 5, 6, 3, 0, -3, 9]可以分成三个区间(2, 4, 8){5, 6}<3, 0, -3, 9>

对于()区间,left=0, right=2, leftmost=true

对于 {}区间, left=3, right=4, leftmost=false,同理可得<>区间的相应参数

当区间长度小于47时,该方法会采用插入排序;否则采用快速排序。
1、 当leftmost为true时,它会采用传统的插入排序(traditional insertion sort),代码也较简单,其过程类似打牌时抓牌插牌。 

2、 当leftmost为false时,它采用一种新型的插入排序(pair insertion sort),改进之处在于每次遍历前面已排好序的数组需要 插入两个元素 ,而传统插入排序在遍历过程中只需要为一个元素找到合适的位置插入。对于插入排序来讲,其关键在于为待插入元素找到合适的插入位置,为了找到这个位置,需要遍历之前已经排好序的子数组,所以对于插入排序来讲,整个排序过程中其遍历的元素个数决定了它的性能。很显然, 每次遍历插入两个元素可以减少排序过程中遍历的元素个数 。

为左边区间时,pair insertion sort在左边元素比较大时,会越界。


双枢轴快速排序

对于快速排序来讲,其每一次递归所做的是使需要排序的子区间变得 更加有序 ,而不是绝对有序;所以对于快速排序来说,其性能决定于每次递归操作使待排序子区间变得有序的程度,另一个决定因素当然就是递归次数。快速排序使子区间变得相对有序的关键是pivot,所以我们优化的方向也应该在于pivot,那就增加pivot的个数吧,而且我们可以发现,增加pivot的个数,对递归次数并不会有太大影响,有时甚至可以使递归次数减少。和insert sort类似的问题就是,pivot增加为几个呢?很显然,pivot的值也不能太大;记住,任何优化都是 有代价的 ,而增加pivot的代价就隐藏在每次 交换 元素的位置过程中。 

下面是寻找枢轴的过程:

// Inexpensive approximation of length / 7,1/7=1/8+1/32
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];


1. pivot的选取方式是将数组分成近视等长的七段,而这七段其实是被5个元素分开的,将这5个元素从小到大排序,取出第2个和第4个,分别作为pivot1和pivot2。

注意:当这个5元素都互不相等时,才采用双枢轴快速排序!

2. Pivot选取完之后,分别从左右两端向中间遍历,左边遍历停止的条件是遇到一个大于等于pivot1的值,并把那个位置标记为less;右边遍历的停止条件是遇到一个小于等于pivot2的值,并把那个位置标记为great

3. 然后从less位置向后遍历,遍历的位置用k表示,会遇到以下几种情况:

a. k位置的值比pivot1小,那就交换k位置和less位置的值,并是less的值加1;这样就使得less位置左边的值都小于pivot1,而less位置和k位置之间的值大于等于pivot1

b. k位置的值大于pivot2,那就从great位置向左遍历,遍历停止条件是遇到一个小于等于pivot2的值,假如这个值小于pivot1,就把这个值写到less位置,把less位置的值写道k位置,把k位置的值写道great位置,最后less++,great--;加入这个值大于等于pivot1,就交换k位置和great位置,之后great--。

4. 完成上述过程之后,带排序的子区间就被分成了三段(pivot2),最后分别对这三段采用递归就行了。
/*
* 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);

就是当中间的数超过4/7的时候,按照划分应该很平均才对,所以猜想中间的元素有很多等于pivot1和pivot2的数(划分的时候等于的数放在中间),会设法减少中间的数,就是把中间的等于pivot1的数放在前方,把等于pivot的数放在后方。

这个做法类似单枢轴快速排序的时候,作为枢轴的元素也有很多相同的,所以在这个时候,应该跳过这些相同元素来进行快速排序。减少递归。

这个做法就相当于再次划分中间的区域,相当于一共根据两个枢轴划分成了5种(多了两种等于p1和p2的)。对其余3种递归。


单枢轴快速排序

当5个元素有相当的时候,假定现在的情况是数组中有很多相同的元素。

} 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);
}

注意:这是个改进的单枢轴快速排序。这个时候也是3个指针的算法。因为,这个算法就像是分成3类,一类小于枢轴,一类大于枢轴,一类等于枢轴。只用对左右两种进行递归。 
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: