1. 排序算法概览
| 算法 | 平均时间 | 最坏时间 | 空间 | 稳定 | 适用场景 |
|---|---|---|---|---|---|
| 冒泡排序 | O(n²) | O(n²) | O(1) | ✅ | 教学/极小数据 |
| 选择排序 | O(n²) | O(n²) | O(1) | ❌ | 极小数据 |
| 插入排序 | O(n²) | O(n²) | O(1) | ✅ | 几乎有序的数据 |
| 希尔排序 | O(n^1.3) | O(n²) | O(1) | ❌ | 中等规模 |
| 归并排序 | O(n log n) | O(n log n) | O(n) | ✅ | 链表排序、外部排序 |
| 快速排序 | O(n log n) | O(n²) | O(log n) | ❌ | 通用场景首选 |
| 堆排序 | O(n log n) | O(n log n) | O(1) | ❌ | 内存敏感、Top-K |
| 计数排序 | O(n + k) | O(n + k) | O(k) | ✅ | 整数范围小 |
| 桶排序 | O(n + k) | O(n²) | O(n + k) | ✅ | 均匀分布数据 |
| 基数排序 | O(d(n + k)) | O(d(n + k)) | O(n + k) | ✅ | 固定位数整数 |
2. 简单排序 O(n²)
2.1 冒泡排序
def bubble_sort(arr):
n = len(arr)
for i in range(n):
swapped = False
for j in range(n - 1 - i):
if arr[j] > arr[j + 1]:
arr[j], arr[j + 1] = arr[j + 1], arr[j]
swapped = True
if not swapped:
break
return arr
2.2 插入排序
def insert_sort(arr):
"""
摸牌式插入,维护已排序前缀
几乎有序时接近 O(n)
"""
for i in range(1, len(arr)):
key = arr[i]
j = i - 1
while j >= 0 and arr[j] > key:
arr[j + 1] = arr[j]
j -= 1
arr[j + 1] = key
return arr
Python 的
list.sort()在数据量小时使用插入排序,大数据用 Timsort(归并+插入的混合)。
3. 高效排序 O(n log n)
3.1 快速排序
分治 + 原地分区。工程中最常用的排序。
import random
def quick_sort(arr, lo=0, hi=None):
if hi is None:
hi = len(arr) - 1
if lo < hi:
p = partition(arr, lo, hi)
quick_sort(arr, lo, p - 1)
quick_sort(arr, p + 1, hi)
return arr
def partition(arr, lo, hi):
"""Lomuto 分区方案"""
pivot = arr[hi]
i = lo - 1
for j in range(lo, hi):
if arr[j] <= pivot:
i += 1
arr[i], arr[j] = arr[j], arr[i]
arr[i + 1], arr[hi] = arr[hi], arr[i + 1]
return i + 1
# 随机化快排(避免最坏情况)
def randomized_partition(arr, lo, hi):
idx = random.randint(lo, hi)
arr[idx], arr[hi] = arr[hi], arr[idx]
return partition(arr, lo, hi)
# 三路快排(处理大量重复元素)
def quick_sort_3way(arr, lo=0, hi=None):
if hi is None:
hi = len(arr) - 1
if lo >= hi:
return
# 分区为 < pivot, == pivot, > pivot
pivot = arr[lo]
lt, i, gt = lo, lo + 1, hi
while i <= gt:
if arr[i] < pivot:
arr[lt], arr[i] = arr[i], arr[lt]
lt += 1
i += 1
elif arr[i] > pivot:
arr[i], arr[gt] = arr[gt], arr[i]
gt -= 1
else:
i += 1
quick_sort_3way(arr, lo, lt - 1)
quick_sort_3way(arr, gt + 1, hi)
return arr
3.2 归并排序
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
# 原地归并(减少空间)
def merge_inplace(arr, start, mid, end):
"""使用临时数组的 O(n) 空间原地归并"""
left = arr[start:mid + 1]
right = arr[mid + 1:end + 1]
i = j = 0
k = start
while i < len(left) and j < len(right):
if left[i] <= right[j]:
arr[k] = left[i]
i += 1
else:
arr[k] = right[j]
j += 1
k += 1
while i < len(left):
arr[k] = left[i]
i += 1
k += 1
while j < len(right):
arr[k] = right[j]
j += 1
k += 1
归并排序是稳定排序,且最坏复杂度保证 O(n log n),适合链表和外部排序。
3.3 堆排序
def heap_sort(arr):
"""原地建堆 + 排序,空间 O(1)"""
n = len(arr)
# 建大顶堆(从最后一个非叶子节点调整)
for i in range(n // 2 - 1, -1, -1):
heapify(arr, n, i)
# 依次将堆顶移到末尾
for i in range(n - 1, 0, -1):
arr[0], arr[i] = arr[i], arr[0]
heapify(arr, i, 0)
return arr
def heapify(arr, n, i):
"""调整以 i 为根的子树为大顶堆"""
largest = i
left = 2 * i + 1
right = 2 * i + 2
if left < n and arr[left] > arr[largest]:
largest = left
if right < n and arr[right] > arr[largest]:
largest = right
if largest != i:
arr[i], arr[largest] = arr[largest], arr[i]
heapify(arr, n, largest)
4. 线性排序
4.1 计数排序
def counting_sort(arr):
"""
适用于整数,范围 [min_val, max_val]
时间 O(n + k),k = 值域大小
"""
if not arr:
return arr
min_val = min(arr)
max_val = max(arr)
count = [0] * (max_val - min_val + 1)
for x in arr:
count[x - min_val] += 1
idx = 0
for i, c in enumerate(count):
while c > 0:
arr[idx] = i + min_val
idx += 1
c -= 1
return arr
4.2 桶排序
def bucket_sort(arr, bucket_count=10):
"""
数据均匀分布在 [0, 1) 时效率最高
每个桶内部使用插入排序
"""
if not arr:
return arr
min_val, max_val = min(arr), max(arr)
buckets = [[] for _ in range(bucket_count)]
for x in arr:
idx = int((x - min_val) / (max_val - min_val) * (bucket_count - 1))
buckets[idx].append(x)
result = []
for bucket in buckets:
result.extend(sorted(bucket)) # 或 insert_sort
return result
4.3 基数排序(LSD)
def radix_sort(arr):
"""从低位到高位排序,稳定"""
if not arr:
return arr
max_val = max(arr)
exp = 1 # 当前位数(个位、十位...)
while max_val // 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 # 0-9 十个桶
# 统计当前位数字频率
for x in arr:
digit = (x // exp) % 10
count[digit] += 1
# 前缀和成为位置索引
for i in range(1, 10):
count[i] += count[i - 1]
# 从后往前填充(保持稳定)
for i in range(n - 1, -1, -1):
digit = (arr[i] // exp) % 10
output[count[digit] - 1] = arr[i]
count[digit] -= 1
for i in range(n):
arr[i] = output[i]
5. 排序算法选择指南
数据规模小 (< 50):
→ 插入排序(常数小,且有适应性)
数据规模中等,随机分布:
→ 快速排序(平均最快)
数据规模大,需要稳定排序:
→ 归并排序 / Timsort
内存极度受限:
→ 堆排序(O(1) 额外空间)
数据是整数且范围小:
→ 计数排序(O(n + k))
数据是固定位数整数:
→ 基数排序(O(d × n))
数据分布均匀:
→ 桶排序(接近 O(n))
已排序/几乎有序:
→ 插入排序(O(n))
6. 稳定性证明示例
稳定的定义:相等元素排序后相对顺序不变。
稳定性重要的场景:对多级排序(如先按年龄排,再按性别排)。
| 算法 | 是否稳定 | 原因 |
|---|---|---|
| 冒泡 | ✅ | 相等不交换 |
| 插入 | ✅ | 相等不后移 |
| 归并 | ✅ | 合并时 left[i] <= right[j] 保证了稳定性 |
| 快排 | ❌ | 分区时跳跃式交换 |
| 堆排 | ❌ | 远距离交换破坏相对顺序 |
7. 并行排序
7.1 多线程快速排序
from concurrent.futures import ThreadPoolExecutor
import threading
def parallel_quick_sort(arr, depth=0, max_depth=4):
"""
递归深度不超过 max_depth 时使用线程池并行
"""
if len(arr) <= 1:
return arr
if depth >= max_depth:
return quick_sort(arr) # 退化为串行
pivot = arr[len(arr) // 2]
left = [x for x in arr if x < pivot]
mid = [x for x in arr if x == pivot]
right = [x for x in arr if x > pivot]
with ThreadPoolExecutor(max_workers=2) as executor:
future_left = executor.submit(parallel_quick_sort, left, depth + 1, max_depth)
future_right = executor.submit(parallel_quick_sort, right, depth + 1, max_depth)
return future_left.result() + mid + future_right.result()
7.2 Java 并行排序
// Arrays.parallelSort():Fork/Join 框架实现
// 阈值:数组长度 < 8192 用双轴快排,否则拆分为多任务
int[] arr = new int[1_000_000];
// 填充随机数据...
Arrays.parallelSort(arr); // 多线程并行排序
// 底层使用 ForkJoinPool.commonPool()
// 时间复杂度:O(n log n),充分利用多核
8. 外部排序(大数据量)
当数据量超过内存容量时,需要使用外部排序:
外部排序流程(归并排序变体):
1. 将大文件切分为多个能放入内存的块(如每块 100MB)
2. 每块读入内存,内部排序后写回临时文件
3. 使用 K 路归并,每次从 K 个文件各读一个缓冲区
4. 将最小元素输出到结果文件,补充对应缓冲区
时间复杂度:O(n log n) + O(n) *(磁盘 I/O 次数)
关键优化:增大缓冲区、减少归并趟数
import heapq
def k_way_merge(files):
"""
K 路归并:使用最小堆从 K 个有序文件中取最小元素
"""
min_heap = []
file_iters = [open(f) for f in files]
# 初始化堆
for i, f in enumerate(file_iters):
line = f.readline().strip()
if line:
heapq.heappush(min_heap, (int(line), i))
with open("sorted_output.txt", "w") as out:
while min_heap:
val, idx = heapq.heappop(min_heap)
out.write(f"{val}\n")
next_line = file_iters[idx].readline().strip()
if next_line:
heapq.heappush(min_heap, (int(next_line), idx))
for f in file_iters:
f.close()
9. TimSort 详解
Python 的 list.sort() 和 Java 的 Arrays.sort() 底层使用 TimSort,一种稳定的混合排序算法:
TimSort 核心思想:
1. 识别数组中的「自然有序段」(run),每个 run 长度 >= min_run(通常 32 或 64)
2. 若 run 长度不足 min_run,使用插入排序扩展到 min_run
3. 使用归并的方式合并相邻 run,保持稳定性
4. 合并策略使用栈和 Galloping 模式优化
适应性:对于部分有序数据,时间接近 O(n)
稳定性:保证相等元素的相对顺序不变
空间:O(n)
| 特性 | TimSort | 快速排序 | 归并排序 |
|---|---|---|---|
| 最坏时间 | O(n log n) | O(n²) | O(n log n) |
| 最好时间 | O(n) | O(n log n) | O(n log n) |
| 稳定性 | ✅ | ❌ | ✅ |
| 适应性 | ✅ 识别有序段 | ❌ | ❌ |
| 空间 | O(n) | O(log n) | O(n) |
10. 排序算法的工程选择
| 语言/库 | 实现 | 特点 |
|---|---|---|
Python sorted() | TimSort | 稳定,适应部分有序 |
Java Arrays.sort() | Dual-Pivot Quicksort(基本类型)/ TimSort(对象) | 基本类型用快排(快),对象用 TimSort(稳定) |
C++ std::sort() | Introsort(快排+堆排+插入) | 保证 O(n log n),最坏触发堆排 |
Rust sort() | Pattern-Defeating Quicksort (pdqsort) | 稳定,适应多种模式 |
Go sort.Slice() | QuickSort + HeapSort fallback | 不稳定,空间 O(log n) |
参考文章
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。