Question

排序是一项非常常用的操作。你的应用程序在运行时，可能无时无刻不在进行排序操作。排序的算法有很多，但在这里我并不打算一一介绍它们。对于大部分排序算法来说，都是串行执行的。当排序元素很多时，若使用并行算法代替串行算法，显然可以更加有效地利用CPU。但将串行算法改造成并行算法并非易事，甚至会极大地增加原有算法的复杂度。在这里，我将介绍几种相对简单的，但是也足以让人脑洞大开的平行排序算法。

5.8.1　分离数据相关性：奇偶交换排序

在介绍奇偶排序前，首先让我们看一下熟悉的冒泡排序。在这里，假设我们需要将数组进行从小到大的排序。冒泡排序的操作很类似水中的起泡上浮，在冒泡排序的执行过程中，如果数据较小，它就会逐步被交换到前面去，相反，对于大的数字，则会下沉，交换到数组的末尾。

冒泡排序的一般算法如下：

01 public static void bubbleSort(int[] arr) {
02   for (int i = arr.length - 1; i ＞ 0; i--) {
03     for (int j = 0; j ＜ i; j++) {
04       if (arr[j] ＞ arr[j + 1]) {
05         int temp = arr[j];
06         arr[j] = arr[j + 1];
07         arr[j + 1] = temp;
08       }
09     }
10   }
11 }

如图5.12所示，展示了冒泡排序的几次迭代过程：

图5.12　冒泡排序迭代过程

大家可以看到，在每次迭代的交换过程中，由于每次交换的两个元素存在数据冲突，对于每个元素，它既可能与前面的元素交换，也可能和后面的元素交换，因此很难直接改造成并行算法。

如果能够解开这种数据的相关性，就可以比较容易地使用并行算法来实现类似的排序。奇偶交换排序就是基于这种思想的。

对于奇偶交换排序来说，它将排序过程分为两个阶段，奇交换和偶交换。对于奇交换来说，它总是比较奇数索引以及其相邻的后续元素。而偶交换总是比较偶数索引和其相邻的后续元素。并且，奇交换和偶交换会成对出现，这样才能保证比较和交换涉及到数组中的每一个元素。

奇偶交换的迭代示意图如图5.13所示。

图5.13　奇偶交换迭代示意图

可以看到，由于将整个比较交换独立分割为奇阶段和偶阶段。这就使得在每一个阶段内，所有的比较和交换是没有数据相关性的。因此，每一次比较和交换都可以独立执行，也就可以并行化了。

下面是奇偶交换排序的串行实现：

01 public static void oddEvenSort(int[] arr) {
02   int exchFlag = 1, start = 0;
03   while (exchFlag == 1 || start == 1) {
04     exchFlag = 0;
05     for (int i = start; i ＜ arr.length - 1; i += 2) {
06       if (arr[i] ＞ arr[i + 1]) {
07         int temp = arr[i];
08         arr[i] = arr[i + 1];
09         arr[i + 1] = temp;
10         exchFlag = 1;
11       }
12     }
13     if (start == 0)
14       start = 1;
15     else
16       start = 0;
17   }
18 }

其中，exchFlag用来记录当前迭代是否发生了数据交换，而start变量用来表示是奇交换还是偶交换。初始时，start为0，表示进行偶交换，每次迭代结束后，切换start的状态。如果上一次比较交换发生了数据交换，或者当前正在进行的是奇交换，循环就不会停止，直到程序不再发生交换，并且当前进行的是偶交换为止（表示奇偶交换已经成对出现）。

上述代码虽然是串行代码，但是已经可以很方便地改造成并行模式：

01 static int exchFlag=1;
02 static synchronized void setExchFlag(int v){
03   exchFlag=v;
04 }
05 static synchronized int getExchFlag(){
06   return exchFlag;
07 }
08
09 public static class OddEvenSortTask implements Runnable{
10   int i;
11   CountDownLatch latch;
12   public OddEvenSortTask(int i,CountDownLatch latch){
13     this.i=i;
14     this.latch=latch;
15   }
16   @Override
17   public void run() {
18     if (arr[i] ＞ arr[i + 1]) {
19       int temp = arr[i];
20       arr[i] = arr[i + 1];
21       arr[i + 1] = temp;
22       setExchFlag(1);
23     }
24     latch.countDown();
25   }
26 }
27 public static void pOddEvenSort(int[] arr) throws InterruptedException {
28   int start = 0;
29   while (getExchFlag() == 1 || start == 1) {
30     setExchFlag(0);
31     //偶数的数组长度，当start为1时，只有len/2-1个线程
32     CountDownLatch latch = new CountDownLatch(arr.length/2-(arr.length%2==0?start:0));
33     for (int i = start; i ＜ arr.length - 1; i += 2) {
34       pool.submit(new OddEvenSortTask(i,latch));
35     }
36     //等待所有线程结束
37     latch.await();
38     if (start == 0)
39       start = 1;
40     else
41       start = 0;
42   }
43 }

上述代码第9行，定义了奇偶排序的任务类。该任务的主要工作是进行数据比较和必要的交换（第18～23行）。并行排序的主体是pOddEvenSort()方法，它使用CountDownLatch记录线程数量，对于每一次迭代，使用单独的线程对每一次元素比较和交换进行操作。在下一次迭代开始前，必须等待上一次迭代所有线程的完成。

5.8.2　改进的插入排序：希尔排序

插入排序也是一种很常用的排序算法。它的基本思想是：一个未排序的数组（当然也可以是链表）可以分为两个部分，前半部分是已经排序的，后半部分是未排序的。在进行排序时，只需要在未排序的部分中选择一个元素，将其插入到前面有序的数组中即可。最终，未排序的部分会越来越少，直到为0，那么排序就完成了。初始时，可以假设已排序部分就是第一个元素。

插入排序的几次迭代示意如图5.14所示。

图5.14　插入排序示意图

插入排序的实现如下所示：

01 public static void insertSort(int[] arr) {
02   int length = arr.length;
03   int j, i, key;
04   for (i = 1; i ＜ length; i++) {
05     //key为要准备插入的元素
06     key = arr[i];
07     j = i - 1;
08     while (j ＞= 0 && arr[j] ＞ key) {
09       arr[j + 1] = arr[j];
10       j--;
11     }
12     //找到合适的位置 插入key
13     arr[j + 1] = key;
14   }
15 }

上述代码第6行，提取要准备插入的元素（也就是未排序序列中的第一个元素）。接着，在已排序队列中找到这个元素的插入位置（第8～10行），并进行插入（第13行）即可。

简单的插入排序是很难并行化的。因为这一次的数据插入依赖于上一次得到的有序序列，因此多个步骤之间无法并行。为此，我们可以对插入排序进行扩展，这就是希尔排序。

希尔排序将整个数组根据间隔h分割为若干个子数组。子数组相互穿插在一起，每一次排序时，分别对每一个子数组进行排序。如图5.15所示，当h为3时，希尔排序将整个数组分为交织在一起的三个子数组。其中，所有的方块为一个子数组，所有的圆形、三角形分别组成另外两个子数组。每次排序时，总是交换间隔为h的两个元素。

图5.15　h=3时的数组分割

在每一组排序完成后，可以递减h的值，进行下轮更加精细的排序。直到h为1，此时等价于一次插入排序。

希尔排序的一个主要优点是，即使一个较小的元素在数组的末尾，由于每次元素移动都以h为间隔进行，因此数组末尾的小元素可以在很少的交换次数下，就被置换到最接近元素最终位置的地方。

下面是希尔排序的串行实现：

01 public static void shellSort(int[] arr) {
02   // 计算出最大的h值
03   int h = 1;
04   while (h ＜= arr.length / 3) {
05     h = h * 3 + 1;
06   }
07   while (h ＞ 0) {
08     for (int i = h; i ＜ arr.length; i++) {
09       if (arr[i] ＜ arr[i - h]) {
10         int tmp = arr[i];
11         int j = i - h;
12         while (j ＞= 0 && arr[j] ＞ tmp) {
13           arr[j + h] = arr[j];
14           j -= h;
15         }
16         arr[j + h] = tmp;
17       }
18     }
19     // 计算出下一个h值
20     h = (h - 1) / 3;
21   }
22 }

上述代码第4～6行，计算一个合适的h值，接着正式进行希尔排序。第8行的for循环进行间隔为h的插入排序，每次排序结束后，递减h的值（第20行）。直到h为1，退化为插入排序。

很显然，希尔排序每次都针对不同的子数组进行排序，各个子数组之间是完全独立的。因此，很容易改写成并行程序：

01 public static class ShellSortTask implements Runnable {
02   int i = 0;
03   int h = 0;
04   CountDownLatch l;
05
06   public ShellSortTask(int i, int h, CountDownLatch latch) {
07     this.i = i;
08     this.h = h;
09     this.l = latch;
10   }
11
12   @Override
13   public void run() {
14     if (arr[i] ＜ arr[i - h]) {
15       int tmp = arr[i];
16       int j = i - h;
17       while (j ＞= 0 && arr[j] ＞ tmp) {
18         arr[j + h] = arr[j];
19         j -= h;
20       }
21       arr[j + h] = tmp;
22     }
23     l.countDown();
24   }
25 }
26
27 public static void pShellSort(int[] arr) throws InterruptedException {
28   // 计算出最大的h值
29   int h = 1;
30   CountDownLatch latch = null;
31   while (h ＜= arr.length / 3) {
32     h = h * 3 + 1;
33   }
34   while (h ＞ 0) {
35     System.out.println("h=" + h);
36     if (h ＞= 4)
37       latch = new CountDownLatch(arr.length - h);
38     for (int i = h; i ＜ arr.length; i++) {
39       // 控制线程数量
40       if (h ＞= 4) {
41         pool.execute(new ShellSortTask(i, h, latch));
42       } else {
43         if (arr[i] ＜ arr[i - h]) {
44           int tmp = arr[i];
45           int j = i - h;
46           while (j ＞= 0 && arr[j] ＞ tmp) {
47             arr[j + h] = arr[j];
48             j -= h;
49           }
50           arr[j + h] = tmp;
51         }
52         // System.out.println(Arrays.toString(arr));
53       }
54     }
55     // 等待线程排序完成，进入下一次排序
56     latch.await();
57     // 计算出下一个h值
58     h = (h - 1) / 3;
59   }
60 }

上述代码中定义ShellSortTask作为并行任务。一个ShellSortTask的作用是根据给定的起始位置和h，对子数组进行排序，因此可以完全并行化。

为控制线程数量，这里定义并行主函数pShellSort()在h大于或等于4时使用并行线程（第40行），否则则退化为传统的插入排序。

每次计算后，递减h的值（第58行）。