10.1 快速、堆与非比较排序
第 10 章 · 排序实现|前置:基础排序、归并排序与二叉堆|下一篇:排序选型。本页实现随机三路快排、堆排、计数、基数与桶排序,并推导比较排序下界。
第二组排序算法的约束
这里没有一种算法包打天下。快速排序拿支点分区,表现取决于分区是否均衡;堆排序提供最坏时间上界与原地核心;计数、基数和桶排序不受比较下界约束,却要读取数据域的额外信息。本章的数学问题是:比较本身究竟设了怎样的速度上限?
先约定接口
下面五个函数都接受任意可迭代对象和可选 key,返回新列表,不修改调用者的输入,且每个元素的 key 只计算一次。为做到这一点,公开包装层会复制并装饰 n 条记录,因此有 Θ(n) 辅助记录和另一个 Θ(n) 输出列表。输入迭代器会被消耗,元素对象按引用共享。原位置只记录身份,不参与比较,不能自动赋予稳定性。快速和堆排序只依赖一致的 < 关系,排除 NaN;下面按求 key、比较和整数运算为常数成本计,任意精度整数的位数成本需另计。后文的“算法辅助空间”指教科书核心过程,不包含这层统一接口。
import random
from math import isfinite
def _decorate(values, key):
key_fn = key if key is not None else lambda item: item
return [[key_fn(item), index, item] for index, item in enumerate(values)]
def _values(records):
return [record[2] for record in records]
def quick_sort(values, key=None, rng=None):
records = _decorate(values, key)
generator = rng if rng is not None else random
def sort_range(low, high):
while low < high:
pivot = records[generator.randrange(low, high + 1)][0]
less = low
scan = low
greater = high
while scan <= greater:
if records[scan][0] < pivot:
records[less], records[scan] = records[scan], records[less]
less += 1
scan += 1
elif pivot < records[scan][0]:
records[scan], records[greater] = records[greater], records[scan]
greater -= 1
else:
scan += 1
if less - low < high - greater:
sort_range(low, less - 1)
low = greater + 1
else:
sort_range(greater + 1, high)
high = less - 1
sort_range(0, len(records) - 1)
return _values(records)
def heap_sort(values, key=None):
records = _decorate(values, key)
def sift_down(root, size):
while 2 * root + 1 < size:
child = 2 * root + 1
if child + 1 < size and records[child][0] < records[child + 1][0]:
child += 1
if not records[root][0] < records[child][0]:
return
records[root], records[child] = records[child], records[root]
root = child
for root in range(len(records) // 2 - 1, -1, -1):
sift_down(root, len(records))
for end in range(len(records) - 1, 0, -1):
records[0], records[end] = records[end], records[0]
sift_down(0, end)
return _values(records)
def counting_sort(values, key=None):
records = _decorate(values, key)
if not records:
return []
if any(not isinstance(record[0], int) for record in records):
raise TypeError("counting_sort requires integer keys")
minimum = min(record[0] for record in records)
maximum = max(record[0] for record in records)
counts = [0] * (maximum - minimum + 1)
for record in records:
counts[record[0] - minimum] += 1
for index in range(1, len(counts)):
counts[index] += counts[index - 1]
output = [None] * len(records)
for record in reversed(records):
slot = record[0] - minimum
counts[slot] -= 1
output[counts[slot]] = record
return _values(output)
def radix_sort(values, key=None):
records = _decorate(values, key)
if not records:
return []
if any(not isinstance(record[0], int) for record in records):
raise TypeError("radix_sort requires integer keys")
offset = min(record[0] for record in records)
largest = max(record[0] - offset for record in records)
place = 1
while place <= largest or place == 1:
counts = [0] * 10
for record in records:
digit = ((record[0] - offset) // place) % 10
counts[digit] += 1
for index in range(1, 10):
counts[index] += counts[index - 1]
output = [None] * len(records)
for record in reversed(records):
digit = ((record[0] - offset) // place) % 10
counts[digit] -= 1
output[counts[digit]] = record
records = output
place *= 10
return _values(records)
def bucket_sort(values, key=None, bucket_count=None):
records = _decorate(values, key)
count = max(1, len(records)) if bucket_count is None else bucket_count
if not isinstance(count, int) or count < 1:
raise ValueError("bucket_count must be a positive integer")
if not records:
return []
if any(not isinstance(record[0], (int, float)) for record in records):
raise TypeError("bucket_sort requires int or float keys")
if any(isinstance(record[0], float) and not isfinite(record[0])
for record in records):
raise ValueError("bucket_sort requires finite keys")
minimum = min(record[0] for record in records)
maximum = max(record[0] for record in records)
if minimum == maximum:
return _values(records)
integer_keys = all(isinstance(record[0], int) for record in records)
if integer_keys:
span = maximum - minimum + 1
else:
try:
span = maximum - minimum
if not isfinite(span):
raise ValueError("floating-point key span must be finite")
except OverflowError as exc:
raise ValueError("floating-point key span is too large") from exc
buckets = [[] for _ in range(count)]
for record in records:
if integer_keys:
index = (record[0] - minimum) * count // span
else:
index = min(count - 1, int((record[0] - minimum) / span * count))
buckets[index].append(record)
for bucket in buckets:
for index in range(1, len(bucket)):
record = bucket[index]
scan = index - 1
while scan >= 0 and record[0] < bucket[scan][0]:
bucket[scan + 1] = bucket[scan]
scan -= 1
bucket[scan + 1] = record
return _values([record for bucket in buckets for record in bucket])复制上方定义后,可用带标签的数据检查输出:
data = [(3, "a"), (-1, "x"), (3, "b"), (0, "y")]
for algorithm in (quick_sort, heap_sort, counting_sort, radix_sort, bucket_sort):
result = algorithm(data, key=lambda item: item[0])
assert [item[0] for item in result] == [-1, 0, 3, 3]
for algorithm in (counting_sort, radix_sort, bucket_sort):
assert algorithm(data, key=lambda item: item[0]) == [
(-1, "x"), (0, "y"), (3, "a"), (3, "b")
]
assert data == [(3, "a"), (-1, "x"), (3, "b"), (0, "y")]快速排序:支点不是保证
对 [4, 2, 4, 1, 4, 3] 选 pivot 4,分区结束后按值属于三个区域,区内顺序可以不同:
< 4: [2, 1, 3] == 4: [4, 4, 4] > 4: []等值区不再参与递归,所以大量重复键不会形成一长串无效分区。随机支点让固定输入上的递归树期望较平衡,因此时间是期望 O(n log n);连续选到极端支点仍可能形成长度 n、n-1、n-2 的子问题,最坏 O(n²)。三路分区改善重复键,不能抹掉这个最坏情况。
循环中维护 [low, less) 小于支点、[less, scan) 等于支点、[scan, greater] 未检查、(greater, high] 大于支点。右端换来的值还没检查,所以交换后只减 greater,不增加 scan。全相等输入只需一趟线性扫描。
支点必须在当前区间中均匀选取,且输入不随随机选择改变,期望界才有对应模型。rng=random.Random(固定种子) 可以复现一次执行;固定种子本身不保证某个输入跑得快。函数先递归较短一侧、循环处理较长一侧,每层递归规模至少减半,因此最坏栈深也为 O(log n),不是仅期望成立。交换会越过相等记录,因此这里的快排不稳定。
堆排序:先把数组变成堆
以 [12, 11, 13, 5, 6, 7] 为例,从最后一个非叶节点索引 2 向前下沉:
root 2: 13 已不小于子节点 7
root 1: 11 已不小于 5、6
root 0: 12 与较大的子节点 13 交换
结果: [13, 11, 12, 5, 6, 7]之后把根 13 与末尾交换,缩短堆,再从根下沉。每次取出当前最大值,已排序区从右向左增长。
为什么建堆是 Θ(n),不是 Θ(n log n)?多数节点靠近叶子,只能下沉很少几层;把每个节点可下沉的高度相加得到线性量。排序阶段执行 n-1 次至多 O(log n) 的下沉,最坏总时间为 Θ(n log n)。每次下沉的上界相加只能直接推出 O(n log n),不能据此声称所有输入都达到该成本;本代码在全相等输入上为 Θ(n)。教科书数组版算法辅助空间 O(1),但本页统一的“不改输入 + key 单次计算”包装仍占 O(n)。堆排序不稳定。
计数排序:范围也是成本
[-2, 1, -2, 0] 的最小键是 -2、最大键是 1,包含两端的范围大小 k = 4。计数数组的槽位 0 到 3 分别代表 -2 到 1。前缀和给出每个键的结束位置,逆序放回让相等记录保持原顺序。
时间 Θ(n + k),额外空间 Θ(n + k)。当 n 只有一千、键却覆盖整个 64 位整数范围时,k 才是真正的瓶颈,计数排序不应使用。此示例会直接按 k 分配数组,没有内存预算保护;接入真实输入前,应先检查范围是否可承受。
基数排序:每一位都必须稳定
本实现先减去最小键,把负数和平移后的非负数放在同一个有序轴上。例如 [-12, 3, -1] 平移成 [0, 15, 11];先按个位稳定排序,再按十位稳定排序,最终顺序与原整数一致。
这里 b=10,从低位到高位处理,称为 LSD 基数排序。设最大平移键的十进制位数为 d,全相等时按一位计,并实际执行一次分配。时间为 Θ(d(n+b)),额外空间为 Θ(n+b)。每趟稳定才能保留前面低位已经确定的顺序;某一趟不稳定,就可能破坏最终结果。
桶排序:平均界来自分布假设
桶排序把数值区间映射到 m 个桶,再用稳定插入排序处理各桶。若第 i 个桶有 n_i 个元素,时间上界为 O(n+m+Σ n_i²)。在独立均匀分桶模型下,平方项的期望为 n+n(n-1)/m,因此期望上界是 O(n+m+n²/m)。只有让负载受控,例如 m=Θ(n),才得到线性期望;m≈√n 时仍可能达到 Θ(n^(3/2))。本实现默认 m=n,也允许调用者指定。可对照 MIT 的桶排序讲解。
输入键限于整数或有限浮点数。全整数使用整数除法分桶,避免大整数转浮点;浮点分支还要求最大值减最小值可表示为有限数,并把最大端点夹回最后一个桶。全相等时直接返回,不会产生二次工作。构造最坏情况应让一个桶包含大量逆序的不同值,例如设置 bucket_count=1。任意 m 下最坏时间上界为 O(n²+m),若 m=O(n²) 才可简写为 O(n²)。
数学关卡:比较排序下界
n 个互异元素有 n! 种排列。只问二元比较的问题时,一个确定性比较排序可画成决策树:内部节点是一次比较,叶子是最终排列。树至少要容纳 n! 个叶子,所以最坏路径长度满足:
因此,比较排序的最坏比较次数下界是 Ω(n log n)。计数和基数排序能突破它,是因为它们读取了整数键范围或数位,不再只靠两两比较。下界没有消失,计算模型变了。
专业速查
| 算法 | 时间 | 稳定 | 教科书算法辅助空间 | 先检查什么 |
|---|---|---|---|---|
| 随机三路快排 | 期望 O(n log n),最坏 O(n²) | 否 | 受控递归 O(log n) | 是否允许不稳定、是否有对抗输入 |
| 堆排序 | 最坏 Θ(n log n),建堆是 Θ(n) | 否 | O(1) | 是否要求最坏界和原地 |
| 计数排序 | Θ(n + k) | 是 | Θ(n + k) | k 是否与 n 同量级 |
| LSD 基数排序 | Θ(d(n + b)) | 是 | Θ(n + b) | 键能否拆成有限数位 |
| 桶排序 | 均匀分桶期望 O(n+m+n²/m),最坏 O(n²+m) | 本实现稳定 | Θ(n + m) | 分布假设能否被测量 |
工程边界
生产代码优先使用语言标准库,再根据基准和约束决定是否自写。Python 文档保证排序稳定,说明 key 每条记录计算一次,也说明实现会利用已有顺序;Java SE 25 文档保证对象数组排序稳定,同时把具体实现写作 implementation note。后者可以随版本变化,不能当成语言永久规则。
来源:
动手验证排序边界
- 用固定随机种子画出
[3, 5, 3, 1, 3, 2]的三路分区。 - 手推七个元素的建堆,记录每次下沉。
- 为计数排序设置
k/n上限,并解释这个上限为什么应由内存预算决定。 - 对互异键的逆序输入设置
bucket_count=1,观察桶内插入排序;再比较全相等输入为何直接返回。
通用排序处理未知数据,计数、基数和桶排序则用更窄的键域或分布合同换速度。快速排序的随机化给出期望界,堆排序给出最坏界与原地核心;这些保证不能互相替代。
完成手算后,去 排序选型把稳定性、内存与输入分布放进同一张决策表。