排序2000万元素时QuickSort卡顿及pivot选择问题排查

问题背景 #

需要对包含id、field1（字符串）、field2、field3的Data结构体数组按field1排序。8000行数据时排序正常，但处理2000万行时程序卡顿；更换pivot为首个或中间元素时，排序结果不正确。

原始代码 #

typedef struct _data
{
    int id;
    char field1[50];
    int field2;
    float field3;
}Data;

typedef struct _GenericArray{
  void **array;
  int size;
  int capacity;
  int (*compare)(void*, void*);
}GenericArray;


void swap(Data* x,Data* y){
     if(x == NULL){
        printf("Can't swap a NULL element(x is NULL)!");
        exit(EXIT_FAILURE);
    }
    if(y == NULL){
        printf("Can't swap a NULL element(y is NULL)!");
        exit(EXIT_FAILURE);
    }
    Data temp = *x;
    *x = *y;
    *y=temp;
}

int partition(GenericArray* generic_array,int first,int last){
    
    if(generic_array == NULL){
        printf("Can't do quickSort in an empty array!");
        exit(EXIT_FAILURE);
    }
    Data *pivot = generic_array->array[last];
    int i = first;
    
    for (int j = first; j < last ;j++)
    {
        if(generic_array->compare(generic_array->array[j],pivot)<=0){
            swap(generic_array->array[i],generic_array->array[j]);
            i++;
        }
    }
    
    swap(generic_array->array[i],generic_array->array[last]);

    return i;
}

int quick_sort(GenericArray* generic_array,int first,int last){
    
    if(generic_array == NULL){
        //printf("Can't do quickSort in an empty array!");
        return -1;
    }
   
    if(first<last){
        int p = partition(generic_array,first,last);
        quick_sort(generic_array,first,p-1);
        quick_sort(generic_array,p+1,last);
    }
    return 0;
}

int compare_string(void* item1, void* item2){  //function in generic_array->compare
    Data *a = (Data*)item1;
    Data *b = (Data*)item2;
    if(strcmp(a->field1,b->field1)==0) return 0;
    else if(strcmp(a->field1,b->field1)>0) return 1;
    else return -1; 
    
}

一、2000万行数据卡顿的核心原因及优化 #

1. 固定pivot导致时间复杂度退化 #

当前partition固定选择最后一个元素作为pivot，当数据接近有序（比如文件读取的内容本身有一定顺序）时，快速排序的划分会极度不平衡，时间复杂度从最优的O(n log n)退化为O(n²)。2000万条数据的O(n²)运算量完全超出可接受的时间范围，直接导致程序卡顿。

2. 结构体全量拷贝的交换开销过大 #

当前swap函数直接拷贝整个Data结构体（包含50字节的字符串字段），每次交换都要复制约62字节的数据。2000万条数据排序过程中会产生大量交换操作，内存拷贝的累计开销会拖慢整体速度。

3. 递归调用的潜在开销 #

标准递归实现的快速排序，在极端不平衡的划分下会产生大量递归调用，虽然2000万数据的最优递归深度仅约25层，但退化场景下的递归次数会大幅增加，进一步加剧卡顿。

优化方案 #

(1) 改用三数取中法选择pivot

通过选择首、中、尾三个元素的中位数作为pivot，避免极端不平衡的划分，保证时间复杂度稳定在O(n log n)：

// 在partition开头添加pivot选择逻辑
int mid = first + (last - first)/2;
// 比较首、中、尾元素，将中位数交换到last位置，复用原有partition逻辑
Data *a = generic_array->array[first];
Data *b = generic_array->array[mid];
Data *c = generic_array->array[last];
if (generic_array->compare(a,b) > 0) swap(&generic_array->array[first], &generic_array->array[mid]);
if (generic_array->compare(a,c) > 0) swap(&generic_array->array[first], &generic_array->array[last]);
if (generic_array->compare(b,c) > 0) swap(&generic_array->array[mid], &generic_array->array[last]);
// 此时last位置是中位数，继续原有逻辑
Data *pivot = generic_array->array[last];

(2) 修改swap为指针交换

由于GenericArray存储的是void**（即Data*的指针数组），直接交换指针而非结构体内容，将交换开销从O(结构体大小)降为O(1)：

void swap(void** x, void** y){
     if(x == NULL || y == NULL){
        printf("Can't swap NULL pointers!");
        exit(EXIT_FAILURE);
    }
    void* temp = *x;
    *x = *y;
    *y = temp;
}
// 调用时传入指针的地址
swap(&generic_array->array[i], &generic_array->array[j]);

(3) 尾递归优化或非递归实现

将递归调用改为尾递归（仅保留一个递归分支），或直接用栈实现非递归快速排序，减少递归调用的栈开销：

// 尾递归优化版quick_sort
int quick_sort(GenericArray* generic_array,int first,int last){
    if(generic_array == NULL) return -1;
   
    while(first<last){
        int p = partition(generic_array,first,last);
        // 优先处理较小的分区，减少递归深度
        if(p - first < last - p){
            quick_sort(generic_array,first,p-1);
            first = p+1;
        } else {
            quick_sort(generic_array,p+1,last);
            last = p-1;
        }
    }
    return 0;
}

二、pivot选首/中间元素时排序失效的原因 #

当前partition函数是基于Lomuto分区法实现的，逻辑依赖于pivot位于last位置：

1. 遍历过程中，所有<=pivot的元素被移到左侧；
2. 最后将pivot（last位置）与i位置交换，完成分区。

如果直接将pivot改为首元素或中间元素，而不调整分区逻辑，会出现以下问题：

• 若pivot在first位置：遍历过程中j从first开始，当交换array[i]和array[j]时，会把pivot移到其他位置，后续比较的pivot还是原来的指针，但该指针指向的元素已经被交换，导致比较逻辑混乱。
• 若pivot在mid位置：遍历结束后交换array[i]和array[last]，但pivot在mid位置未被移动到正确的分区点，导致分区结果错误，最终排序失败。

修复方法 #

无论选择哪个位置作为pivot，都需要先将其交换到last位置，再复用原有的Lomuto分区逻辑，比如选择中间元素作为pivot：

int partition(GenericArray* generic_array,int first,int last){
    if(generic_array == NULL){
        printf("Can't do quickSort in an empty array!");
        exit(EXIT_FAILURE);
    }
    // 选择中间元素作为pivot，交换到last位置
    int mid = first + (last - first)/2;
    swap(&generic_array->array[mid], &generic_array->array[last]);
    Data *pivot = generic_array->array[last];
    int i = first;
    
    for (int j = first; j < last ;j++)
    {
        if(generic_array->compare(generic_array->array[j],pivot)<=0){
            swap(&generic_array->array[i], &generic_array->array[j]);
            i++;
        }
    }
    
    swap(&generic_array->array[i], &generic_array->array[last]);
    return i;
}

内容的提问来源于stack exchange，提问作者Matteo Pagliarello