高级排序算法：从快排到外部排序

引言

排序是计算机科学中最基础也是最重要的操作之一。虽然日常开发中我们常使用语言内置的排序函数，但理解底层算法原理对于优化性能和解决特殊场景问题至关重要。本文将深入讲解高级排序算法及其工程应用。

一、快速排序的深度优化

1.1 基础快速排序

def quick_sort(arr):
    if len(arr) <= 1:
        return arr
    
    pivot = arr[len(arr) // 2]
    left = [x for x in arr if x < pivot]
    middle = [x for x in arr if x == pivot]
    right = [x for x in arr if x > pivot]
    
    return quick_sort(left) + middle + quick_sort(right)
​

时间复杂度： 平均O(n log n)，最坏O(n²)

1.2 原地快速排序

def quick_sort_inplace(arr, low, high):
    if low < high:
        pivot_index = partition(arr, low, high)
        quick_sort_inplace(arr, low, pivot_index - 1)
        quick_sort_inplace(arr, pivot_index + 1, high)

def partition(arr, low, high):
    pivot = arr[high]
    i = low - 1
    
    for j in range(low, high):
        if arr[j] <= pivot:
            i += 1
            arr[i], arr[j] = arr[j], arr[i]
    
    arr[i + 1], arr[high] = arr[high], arr[i + 1]
    return i + 1
​

1.3 三路快排（处理重复元素）

def quick_sort_3way(arr, low, high):
    if low >= high:
        return
    
    lt, gt = low, high
    i = low + 1
    pivot = arr[low]
    
    while i <= gt:
        if arr[i] < pivot:
            arr[lt], arr[i] = arr[i], arr[lt]
            lt += 1
            i += 1
        elif arr[i] > pivot:
            arr[gt], arr[i] = arr[i], arr[gt]
            gt -= 1
        else:
            i += 1
    
    quick_sort_3way(arr, low, lt - 1)
    quick_sort_3way(arr, gt + 1, high)
​

1.4 随机化与优化

import random

def quick_sort_optimized(arr, low, high):
    if low < high:
        # 小数组使用插入排序
        if high - low < 16:
            insertion_sort(arr, low, high)
            return
        
        # 随机选择pivot
        pivot_index = random.randint(low, high)
        arr[pivot_index], arr[high] = arr[high], arr[pivot_index]
        
        pivot_index = partition(arr, low, high)
        quick_sort_optimized(arr, low, pivot_index - 1)
        quick_sort_optimized(arr, pivot_index + 1, high)

def insertion_sort(arr, low, high):
    for i in range(low + 1, high + 1):
        key = arr[i]
        j = i - 1
        while j >= low and arr[j] > key:
            arr[j + 1] = arr[j]
            j -= 1
        arr[j + 1] = key
​

二、归并排序与外部排序

2.1 归并排序

def merge_sort(arr):
    if len(arr) <= 1:
        return arr
    
    mid = len(arr) // 2
    left = merge_sort(arr[:mid])
    right = merge_sort(arr[mid:])
    
    return merge(left, right)

def merge(left, right):
    result = []
    i = j = 0
    
    while i < len(left) and j < len(right):
        if left[i] <= right[j]:
            result.append(left[i])
            i += 1
        else:
            result.append(right[j])
            j += 1
    
    result.extend(left[i:])
    result.extend(right[j:])
    return result
​

2.2 外部排序（处理大数据）

当数据量超过内存容量时，需要使用外部排序：

import heapq

def external_sort(input_file, output_file, chunk_size=10000):
    # 第一阶段：分块排序
    chunks = []
    with open(input_file, 'r') as f:
        chunk = []
        for line in f:
            chunk.append(int(line.strip()))
            if len(chunk) >= chunk_size:
                chunk.sort()
                chunks.append(chunk)
                chunk = []
        
        if chunk:
            chunk.sort()
            chunks.append(chunk)
    
    # 第二阶段：多路归并
    with open(output_file, 'w') as out:
        # 使用堆进行多路归并
        heap = []
        for i, chunk in enumerate(chunks):
            if chunk:
                heapq.heappush(heap, (chunk[0], i, 0))
        
        while heap:
            val, chunk_idx, elem_idx = heapq.heappop(heap)
            out.write(f"{val}\n")
            
            if elem_idx + 1 < len(chunks[chunk_idx]):
                next_val = chunks[chunk_idx][elem_idx + 1]
                heapq.heappush(heap, (next_val, chunk_idx, elem_idx + 1))
​

三、特殊场景排序

3.1 计数排序（整数排序）

def counting_sort(arr):
    if not arr:
        return arr
    
    max_val = max(arr)
    min_val = min(arr)
    range_val = max_val - min_val + 1
    
    count = [0] * range_val
    output = [0] * len(arr)
    
    # 统计每个元素出现次数
    for num in arr:
        count[num - min_val] += 1
    
    # 计算累积位置
    for i in range(1, len(count)):
        count[i] += count[i - 1]
    
    # 从后向前构建输出（保持稳定性）
    for i in range(len(arr) - 1, -1, -1):
        output[count[arr[i] - min_val] - 1] = arr[i]
        count[arr[i] - min_val] -= 1
    
    return output
​

时间复杂度： O(n + k)，k为数据范围

3.2 基数排序

def radix_sort(arr):
    if not arr:
        return arr
    
    max_num = max(arr)
    exp = 1
    
    while max_num // exp > 0:
        counting_sort_by_digit(arr, exp)
        exp *= 10
    
    return arr

def counting_sort_by_digit(arr, exp):
    n = len(arr)
    output = [0] * n
    count = [0] * 10
    
    for i in range(n):
        index = (arr[i] // exp) % 10
        count[index] += 1
    
    for i in range(1, 10):
        count[i] += count[i - 1]
    
    for i in range(n - 1, -1, -1):
        index = (arr[i] // exp) % 10
        output[count[index] - 1] = arr[i]
        count[index] -= 1
    
    for i in range(n):
        arr[i] = output[i]
​

3.3 桶排序

def bucket_sort(arr, bucket_size=5):
    if not arr:
        return arr
    
    min_val, max_val = min(arr), max(arr)
    bucket_count = (max_val - min_val) // bucket_size + 1
    buckets = [[] for _ in range(bucket_count)]
    
    # 分配到桶
    for num in arr:
        index = (num - min_val) // bucket_size
        buckets[index].append(num)
    
    # 对每个桶排序
    result = []
    for bucket in buckets:
        bucket.sort()
        result.extend(bucket)
    
    return result
​

四、工程应用场景

4.1 数据库排序

数据库系统使用复杂的排序策略：

• 小数据集：使用快速排序
• 大数据集：使用外部归并排序
• 有序数据：使用归并排序

4.2 分布式排序

# MapReduce风格的分布式排序
def map_reduce_sort(data, mappers, reducers):
    # Map阶段：分片
    chunks = [data[i::mappers] for i in range(mappers)]
    
    # 本地排序
    sorted_chunks = [sorted(chunk) for chunk in chunks]
    
    # Shuffle阶段：按范围分配
    ranges = partition_ranges(sorted_chunks, reducers)
    reducer_inputs = [[] for _ in range(reducers)]
    
    for chunk in sorted_chunks:
        for item in chunk:
            reducer_idx = find_reducer(item, ranges)
            reducer_inputs[reducer_idx].append(item)
    
    # Reduce阶段：归并
    final_results = []
    for input_chunk in reducer_inputs:
        final_results.extend(sorted(input_chunk))
    
    return final_results
​

五、性能对比与选择

总结

选择排序算法需要考虑：

1. 数据规模：小数组用插入排序，大数组用快排/归并
2. 数据特性：部分有序用插入排序，整数用计数排序
3. 内存限制：内存不足用外部排序
4. 稳定性要求：需要稳定排序用归并排序

掌握这些算法，你就能在各种场景下选择最优的排序方案。

扩展阅读

• TimSort：Python和Java使用的混合排序算法
• IntroSort：快速排序、堆排序和插入排序的混合
• 排序网络：并行排序算法
• GPU加速排序