跳转至
计算机基础手册 数据结构与算法

排序、堆与贪心：先排序，再一遍扫过去

这一类题的解法常常只有两行：排个序，然后扫一遍。难的是想清楚"按什么排"和"为什么这么选是对的"。堆则是"动态取最值"的标准工具——Top-K 采样、多路归并、调度器每次挑最早空闲的机器，用的都是它。这一章把 Top-K、多路归并、区间问题、任务调度这几个模板讲清楚，并对照推理系统里的排序与装箱问题。

自测：能答上来就可以跳过本章

  1. 求 n 个数里最大的 k 个，用排序和用堆各是什么复杂度？什么时候该用哪个？
  2. Python 的 heapq 是小顶堆，要大顶堆怎么办？堆里要放元组时要注意什么？
  3. 合并重叠区间按什么排序？"最多选几个互不重叠的区间"又按什么排序？
  4. sorted 的稳定性是什么意思？多关键字排序怎么写？
  5. 贪心为什么需要"证明"？举一个贪心不对的例子。
自测参考答案（先自己答，再展开对照）
  1. 排序是 O(n log n)，维护大小为 k 的小顶堆是 O(n log k)。k 远小于 n 时（比如一亿条日志里取前 100）用堆；k 接近 n、或者后面还要用整体有序的结果时直接排序。还有一个 O(n) 平均的快速选择（quickselect），只要第 k 大而不要顺序时最快。
  2. 把值取负（heappush(h, -x)），或者放元组 (-优先级, 数据)。元组比较是逐项的，所以第二项必须可比较；放对象时要加一个单调递增的序号打破平局，否则会去比较对象本身而报 TypeError。
  3. 合并重叠区间按起点排序；最多选不重叠的区间按终点排序（越早结束越好，给后面留的空间最多）。按起点选最多是错的。
  4. 稳定是指排序不改变相等元素之间的原有顺序。多关键字直接写 key=lambda x: (-x[1], x[0])（先按第二项降序、再按第一项升序），或者利用稳定性排两次（先排次关键字，再排主关键字）。
  5. 因为贪心每一步都做"当前最优"的选择，不一定导致全局最优。要么给出交换论证（"任何最优解都可以改造成包含我这一步选择的解"），要么举反例。经典反例：硬币面额 [1, 3, 4] 凑 6，贪心先拿 4 再拿两个 1 共 3 枚，最优是两个 3 共 2 枚——所以零钱兑换要用动态规划。

堆：动态取最值

3 [0] 7 [1] 5 [2] 12 [3] 9 [4] 8 [5] 15 [6] 数组：[3, 7, 5, 12, 9, 8, 15] 最小堆：每个父节点 ≤ 两个孩子；没有指针，缓存友好 push(4)：放到末尾 [7]，和父节点 [3]=12 比，小就交换，一路上浮到不比父节点小为止：O(log n) pop()：取走堆顶 3，把末尾的 15 放到顶上，和两个孩子中较小的交换，一路下沉：O(log n) heapify：从最后一个非叶子节点倒着下沉，O(n)，比逐个 push 的 O(n log n) 快 用在哪：Top-K（维护大小为 K 的堆）、合并 K 路有序流、按截止时间调度（推理引擎按预计完成时间取请求就是一个堆）
图：堆——用数组存的完全二叉树，上浮与下沉
heaps.py
# 堆的三种典型用法：Top-K、多路归并、以及"动态取最小"的调度
import heapq
import random

rng = random.Random(0)
data = [rng.randrange(1000) for _ in range(20000)]

# 1. Top-K：维护一个大小为 k 的小顶堆，堆顶是"第 k 大"
def top_k(a, k):
    heap = []
    for x in a:
        if len(heap) < k:
            heapq.heappush(heap, x)
        elif x > heap[0]:                      # 比第 k 大还大，替换堆顶
            heapq.heapreplace(heap, x)
    return sorted(heap, reverse=True)


print("Top-5：", top_k(data, 5), "（和排序取前 5 一致：", sorted(data, reverse=True)[:5] == top_k(data, 5), "）")
print("复杂度：排序 O(n log n)，堆 O(n log k)——k 远小于 n 时堆明显更省")

# 2. 多路归并：k 条有序序列合成一条
streams = [sorted(rng.randrange(100) for _ in range(5)) for _ in range(4)]
print("\n四条有序序列：", streams)
print("归并结果：", list(heapq.merge(*streams)))

# 3. 调度：每次取"最早空闲"的机器，就是一个小顶堆
def schedule(tasks, machines):
    """tasks 是每个任务的耗时，返回全部完成的时刻"""
    heap = [0.0] * machines                    # 每台机器的空闲时刻
    heapq.heapify(heap)
    for cost in tasks:
        free_at = heapq.heappop(heap)          # 最早空闲的机器
        heapq.heappush(heap, free_at + cost)
    return max(heap)


tasks = [5, 3, 8, 2, 7, 1, 6]
for m in (1, 2, 3):
    print(f"{len(tasks)} 个任务（共 {sum(tasks)} 单位）用 {m} 台机器：{schedule(tasks, m):.0f} 单位完成，"
          f"理论下界 {max(max(tasks), sum(tasks) / m):.1f}")
print("\n把任务按耗时从大到小排序再调度（最长优先，LPT）：")
for m in (2, 3):
    print(f"  {m} 台：顺序到达 {schedule(tasks, m):.0f}，先排序 {schedule(sorted(tasks, reverse=True), m):.0f}")
输出
Top-5： [999, 999, 999, 999, 999] （和排序取前 5 一致： True ）
复杂度：排序 O(n log n)，堆 O(n log k)——k 远小于 n 时堆明显更省

四条有序序列： [[28, 29, 46, 69, 71], [1, 6, 6, 56, 68], [33, 48, 51, 52, 66], [2, 26, 39, 81, 94]]
归并结果： [1, 2, 6, 6, 26, 28, 29, 33, 39, 46, 48, 51, 52, 56, 66, 68, 69, 71, 81, 94]
7 个任务（共 32 单位）用 1 台机器：32 单位完成，理论下界 32.0
7 个任务（共 32 单位）用 2 台机器：18 单位完成，理论下界 16.0
7 个任务（共 32 单位）用 3 台机器：12 单位完成，理论下界 10.7

把任务按耗时从大到小排序再调度（最长优先，LPT）：
  2 台：顺序到达 18，先排序 16
  3 台：顺序到达 12，先排序 11

三个用法各对应一类题：

  • Top-K：维护大小为 k 的小顶堆，堆顶就是"当前的第 k 大"。比它小的直接丢弃。推理里的对应是采样时的 top-k（词表 15 万里取前 50），GPU 上用的是并行归约版本，思路一样。
  • 多路归并：k 条有序序列各放一个"当前元素"进堆，每次取最小的、补上它的后继。外排序、合并 k 个有序链表、分布式查询的结果归并都是它。Python 的 heapq.merge 直接可用。
  • 调度：堆里放"每台机器的空闲时刻"，每次取最早空闲的。这就是推理调度器的雏形——不过真实调度还要考虑 KV 显存、前缀命中、优先级（见调度器）。

heapq 的两个坑：只有小顶堆（要大顶堆就取负）；元组比较会逐项进行，所以放 (优先级, 对象) 时要在中间插一个序号，否则优先级相同时会去比较对象本身。

排序：稳定性与自定义键

Python 的 sorted 是 Timsort：最坏 O(n log n)，对"部分有序"的数据接近 O(n)，并且稳定。稳定性让多关键字排序有两种写法：

items.sort(key=lambda x: (-x.score, x.name))        # 一次排完：分数降序、同分按名字
items.sort(key=lambda x: x.name)                     # 或者排两次，先次要后主要
items.sort(key=lambda x: -x.score)

几个实用的点：

  • 不要写 cmp 风格的比较函数，Python 3 只接受 key；实在需要就用 functools.cmp_to_key。
  • 按多个字段排序时用元组，不要拼字符串。
  • 大数据量只要前 k 个：heapq.nlargest(k, data) 比 sorted(data)[:k] 省内存。
  • 快速选择（quickselect）平均 O(n) 找第 k 大，最坏 O(n²)（随机选枢轴可以避免）。面试问"O(n) 找第 k 大"时说的就是它。

贪心：排序 + 一遍扫描

greedy.py
# 贪心：先按某个关键字排序，再一遍扫过去。难的不是代码，是证明"为什么这么选是对的"
def merge_intervals(intervals):
    """合并重叠区间：按起点排序，能接上就扩展右端点"""
    out = []
    for start, end in sorted(intervals):
        if out and start <= out[-1][1]:
            out[-1][1] = max(out[-1][1], end)
        else:
            out.append([start, end])
    return out


def max_non_overlapping(intervals):
    """最多能选几个互不重叠的区间：按**终点**排序，能选就选"""
    count, last_end = 0, float("-inf")
    for start, end in sorted(intervals, key=lambda t: t[1]):
        if start >= last_end:
            count += 1
            last_end = end
    return count


def min_rooms(meetings):
    """同时进行的最大数量（需要几个会议室 / 几个并发槽位）：把端点排序后扫描"""
    events = sorted([(s, 1) for s, _ in meetings] + [(e, -1) for _, e in meetings])
    cur = best = 0
    for _, delta in events:                    # 同一时刻先处理结束（-1 排在 +1 前面）
        cur += delta
        best = max(best, cur)
    return best


data = [[1, 3], [2, 6], [8, 10], [15, 18]]
print("区间：", data)
print("合并后：", merge_intervals(data))
print("最多能选几个互不重叠：", max_non_overlapping(data))
print("需要几个并发槽位：", min_rooms(data))
print()
print("三道题排序的关键字不同：合并按起点，选最多按终点，算并发按端点扫描。")
print("按起点选最多是错的——先来的可能很长，挡住后面好几个短的。")
print()

# 推理系统里的例子：按输出长度排序能减少"批次里的浪费"
reqs = [("A", 500), ("B", 20), ("C", 480), ("D", 30)]
for name, order in [("按到达顺序", reqs), ("按预估长度分组", sorted(reqs, key=lambda t: t[1]))]:
    batches = [order[:2], order[2:]]
    waste = sum(max(x[1] for x in b) * len(b) - sum(x[1] for x in b) for b in batches)
    print(f"{name}：分两批 {[[x[0] for x in b] for b in batches]}，浪费的 token 槽位 {waste}")
输出
区间： [[1, 3], [2, 6], [8, 10], [15, 18]]
合并后： [[1, 6], [8, 10], [15, 18]]
最多能选几个互不重叠： 3
需要几个并发槽位： 2

三道题排序的关键字不同：合并按起点，选最多按终点，算并发按端点扫描。
按起点选最多是错的——先来的可能很长，挡住后面好几个短的。

按到达顺序：分两批 [['A', 'B'], ['C', 'D']]，浪费的 token 槽位 930
按预估长度分组：分两批 [['B', 'D'], ['C', 'A']]，浪费的 token 槽位 30

贪心题的解法通常就是"按某个关键字排序，然后一遍扫过去"，难点全在按什么排：

题型 排序关键字 直觉
合并重叠区间 起点 能接上就扩展右端点
最多不重叠区间 / 最少箭数 终点 越早结束，给后面留的空间越多
会议室数量 / 最大并发 端点扫描（差分） 进一个 +1、出一个 -1
最少平台数、任务调度 堆 + 时间顺序 每次复用最早空闲的资源
分发饼干、分配任务 两个数组都排序 双指针配对
加油站、跳跃游戏 一遍扫描维护"能到的最远处" 局部可达性

最后那个例子是推理系统里的真实问题：同一批请求的输出长度差异越大，浪费越多。一个批次要跑到最长的那个请求结束，短请求的槽位就空着（连续批处理会及时替换掉完成的请求，但一步之内仍然是对齐的）。所以按预估长度分组、或者把长请求单独成批，能明显提高吞吐——这和"装箱问题"是同一类贪心。大规模 MoE 部署里的专家负载均衡（EPLB）也是装箱：把专家分配到卡上，使最重的卡尽量轻（见大规模 EP 部署）。

贪心要么能证明，要么就是错的。 常用的证明手法是交换论证：假设存在一个最优解和我的选择不同，把它换成我的选择后不会变差。证不出来就老实用动态规划——零钱兑换（面额 [1, 3, 4] 凑 6）就是贪心翻车的经典例子。

这一章的题在考什么

看到这些字眼 往哪想
第 k 大、前 k 个、频率最高 堆（O(n log k)）或快速选择（O(n)）
合并 k 个有序、外排序 多路归并（堆）
区间、重叠、合并、最少删除 排序 + 贪心（注意按起点还是终点）
会议室、并发数、最大重叠 端点排序扫描 / 差分
任务分配、最早完成、最少机器 堆模拟 + 贪心（LPT）
装箱、负载均衡、分组 贪心近似（最大剩余、最长优先）
自定义排序、多关键字 key=lambda x: (-a, b)

怎么讲清楚

Top-K 先给三种解法和复杂度："排序 O(n log n)、大小为 k 的堆 O(n log k)、快速选择平均 O(n)"，再说场景（数据流只能用堆、要全序就排序）。讲贪心必须主动说清楚按什么排序和为什么：合并区间按起点、选最多不重叠按终点（交换论证：越早结束越好）。heapq 的细节（只有小顶堆、元组里要加序号防止比较对象）是很好的加分点。能把题目映射到工程会更好：Top-K 对应采样、多路归并对应外排序与分布式结果合并、装箱对应专家负载均衡、按输出长度分组对应减少批次里的浪费。最后别忘了说"贪心要能证明，证不出来就上动态规划"。

小结

  • Top-K 用大小为 k 的小顶堆 O(n log k)，只要第 k 大可以用快速选择 O(n) 平均。
  • heapq 只有小顶堆（取负得大顶堆）；元组比较要加序号防止比较对象本身。
  • 多路归并用堆，是外排序和分布式结果合并的基础。
  • 贪心 = 排序 + 一遍扫描；合并区间按起点、选最多不重叠按终点、算并发用端点扫描。
  • 调度用堆取"最早空闲"，装箱用最长优先；推理里对应请求分组与专家负载均衡。
  • 贪心要能证明（交换论证），证不出来就用动态规划。

本章练习题（8 道）

在浏览器里直接写代码、跑测试；也可以在本地用 python practice/judge.py 判题。

发现错误或有更好的写法？在 GitHub 上编辑此页提交问题