Python heapq.merge 怎样合并多个有序输入流

需要把多个已经排好序的日志流、文件行流或数据库导出流拼成一个有序结果时，可以使用 heapq.merge()。它返回的是惰性迭代器，不会像 sorted(itertools.chain(...)) 那样先把所有输入读入内存；但它有一个必须遵守的前提：每个输入都要按同一比较规则独立有序。

最小写法是 heapq.merge(stream_a, stream_b, stream_c)。默认按从小到大合并；如果输入是记录对象，用关键字参数 key 指定比较字段；如果要降序，所有输入先按降序排列，再传入 reverse=True。

官方地址：https://docs.python.org/3.14/library/heapq.html

heapq.merge 解决的不是“重新排序”

heapq.merge 更像多个有序输入的归并器，而不是一个接收任意数据后再排序的函数。假设三个输入分别是 [1, 4, 9]、[2, 3, 8] 和 [0, 5, 7]，每个输入内部已经从小到大排列，合并后才会得到完整的升序序列。

与整体排序相比，归并的优势在于输入可以是生成器、文件迭代器或其他只支持逐项读取的对象。调用函数本身不会立刻消耗所有数据，真正遍历结果时才会继续向输入取值。

用生成器合并多个有序输入流

下面的示例把三个生成器作为输入，并只消费前五个结果。代码中的生成器会在需要时产生数据，适合演示“结果迭代到哪里，输入读取到哪里”。

from heapq import merge


def numbers(start, stop, step):
    """产生一个已经按升序排列的整数流。"""
    current = start
    while current = 7:
        break

这个例子会按升序逐项产生结果。break 之后没有继续消费迭代器，因此尚未读取的输入也不会被提前展开。若换成文件对象，调用方式相同，只需保证每个文件已经按同一个排序规则排列。

用 key 按记录字段合并

实际数据通常不是整数，而是字典、元组或自定义对象。此时可以使用关键字参数 key 提取比较字段，输出的仍然是完整记录，而不是提取出来的字段。

from heapq import merge


left = [
    {"time": 2, "source": "A", "message": "connected"},
    {"time": 8, "source": "A", "message": "closed"},
]
right = [
    {"time": 3, "source": "B", "message": "accepted"},
    {"time": 6, "source": "B", "message": "handled"},
]

# 两个输入都已按 time 升序；key 只负责告诉 merge 比较哪个字段。
merged = merge(left, right, key=lambda record: record["time"])

for record in merged:
    # 下游仍能拿到 source 和 message 等完整字段。
    print(record["time"], record["source"], record["message"])

这里不要写成 merge(left, right, lambda record: ...)，因为 key 和 reverse 是关键字参数。还要注意，key 只改变比较方式，不会替你重新整理每个输入；如果 left 或 right 按 time 不是升序，结果也不会自动修正。

reverse=True 要求所有输入都是降序

降序合并时，不能只把 reverse=True 当成“对最终结果反转”。官方文档要求所有输入也按从大到小排列，合并器才会在相同方向上比较。

from heapq import merge


first = [{"time": 9}, {"time": 4}, {"time": 1}]
second = [{"time": 8}, {"time": 6}, {"time": 2}]

# 每个输入都按 time 降序排列，reverse=True 才与输入方向一致。
merged = merge(first, second, key=lambda record: record["time"], reverse=True)

for record in merged:
    # 输出顺序为 9、8、6、4、2、1。
    print(record["time"])

如果输入仍然是升序，却传入 reverse=True，就违反了函数的输入前提，结果不能按预期排列。工程代码中应把“输入排序方向”和“merge 参数”作为同一份配置维护，避免只改一边。

重复值、空输入和无限流怎么处理

多个输入出现相同值时，heapq.merge 会保留这些值，不会自动去重。因此它适合做日志或增量数据的归并，不适合直接当作去重工具。若要去重，需要在消费结果时比较相邻值，并明确重复记录的业务规则。

空列表、空生成器都可以作为输入，不会单独报错；如果所有输入都为空，结果就是一个空迭代器。输入数量也可以是零，list(merge()) 会得到空列表。需要注意的是，若某个输入是无限生成器，只有在下游也有限制消费数量时才适合使用，否则遍历结果不会自然结束。

什么时候不应该使用 heapq.merge

如果原始输入没有分别排序，先用 merge 并不能得到全局有序结果。这时应先对每个分片排序，再进行归并；或者数据量允许时直接使用整体排序。两者的选择可以按下面的条件判断：

最后记住这四个判断

• 每个输入先独立排序，默认方向是从小到大。
• heapq.merge 返回惰性迭代器，消费结果时才继续读取输入。
• 记录对象使用 key= 指定比较字段，输出仍保留完整记录。
• reverse=True 只适用于所有输入都按降序排列的情况；重复值不会被自动删除。

常见问题

heapq.merge 会把全部输入读进内存吗？

不会一次性读入全部输入，它按需返回迭代器。但它仍会维护每个输入的当前候选值，因此“流式”不等于完全没有内存开销。

heapq.merge 能合并字典列表吗？

可以，给出 key 函数即可；前提是每个字典列表都已经按相同的 key 规则排序。

为什么 reverse=True 后结果还是不对？

最常见原因是输入仍按升序排列。先把每个输入按比较字段降序排列，再把同一个字段传给 key，最后使用 reverse=True。

heapq.merge 会自动去重吗？

不会。相同值会按输入中的记录继续输出；去重必须由下游根据值或业务主键另行处理。