Python heapq 稳定优先队列:用序号处理并列,避免比较任务内容

10-01 5阅读

先约定“同样着急”时谁先执行

假设一个图片处理脚本收到普通任务甲、紧急任务、普通任务乙和普通任务丙。业务规则是紧急任务先取,三个普通任务按到达次序取。任务本身是字典,包含名称、路径和处理参数;这些字段用于执行任务,不能顺便成为排序规则。先写出期望顺序,才知道优先队列需要保证什么。

标准库 heapq 默认维护最小堆,数值较小的优先级先取出。官方文档指出,两项元组在优先级相同时会继续比较任务;字典无法这样排序。把条目改成“优先级、唯一序号、任务”后,比较会在序号处分出先后,不必触碰任务内容。这里用 count 生成递增整数。

Python heapq 稳定优先队列:用序号处理并列,避免比较任务内容

AI生成概念示意图,非真实界面

让错误样本和正确样本同时出现

下面代码可直接用 Python 三运行。前半段故意压入两个同优先级字典,捕获比较错误后丢弃这个试验堆。后半段建立新的队列,只接受整数优先级,把规则限制在容易核验的范围。序号属于整个队列实例,每次入队只领取一次;它不取自任务名称,也不使用可能重复的秒级时间戳。

from heapq import heappush, heappop
from itertools import count

broken = []
heappush(broken, (2, {"name": "A"}))
try:
    heappush(broken, (2, {"name": "B"}))
except TypeError:
    print("two-field entry: TypeError")
else:
    raise AssertionError("expected incomparable payloads")
# Do not reuse a heap after a comparison failure.

class StableQueue:
    def __init__(self):
        self._heap = []
        self._serial = count()

    def push(self, priority, payload):
        if type(priority) is not int:
            raise TypeError("priority must be an integer")
        heappush(self._heap, (priority, next(self._serial), payload))

    def pop(self):
        return heappop(self._heap)[2]

q = StableQueue()
for priority, name in [(2, "A"), (1, "urgent"), (2, "B"), (2, "C")]:
    q.push(priority, {"name": name})
names = [q.pop()["name"] for _ in range(4)]
assert names == ["urgent", "A", "B", "C"]
print("stable order:", names)
try:
    q.pop()
except IndexError:
    print("empty queue: IndexError")
else:
    raise AssertionError("expected empty queue failure")
payload = object()
q.push(0, payload)
assert q.pop() is payload
print("opaque payload: preserved")

结果应先报告两项条目的类型错误,再输出紧急任务以及甲、乙、丙的顺序。代码使用英文短名方便对照。最后还取出一个没有业务字段的对象,验证返回的是原对象。这个检查提醒我们,队列不需要理解任务结构;给任意任务添加比较方法,会把调度策略混入数据模型,后续调整更难定位。

稳定有范围,堆内列表不是执行日志

这里的稳定只针对优先级相等的条目。紧急任务仍然可以越过先来的普通任务;如果紧急工作持续涌入,普通工作也可能长期等待。序号没有解决公平性,更没有规定完成顺序。多个执行者拿到任务后,慢任务可能最后结束,因此页面若显示“完成先后”,不能直接复制这里的入队顺序。

调试时应该逐个弹出检查,不要把内部列表从头到尾打印后当成完整排序结果。也不要直接修改已入堆条目的优先级,再期待下一次弹出自动修复全部关系。需要取消或调级时,另行设计失效标记与重新入队规则,并约定重新入队是否领取新序号;这个决定会改变并列任务之间的公平关系。

把检查集中在并列边界

人工验收时,先让所有任务优先级相同,核对每个名字是否保持顺序;再插入一个更紧急任务,检查它能否提前。随后试一个元素和空队列,确认调用者能够处理空队列异常。真实输入若来自文本配置,应在入队前统一转换和校验,别让数字字符串与整数混在一次比较里。

这一实现面向单线程脚本。在线程间传递工作时,可查看官方同步队列接口,并继续明确并列项的排序规则。尤其要先定义“到达”是请求收到、任务创建还是成功入队:三者在忙碌系统中未必相同。测试中的序号证明的是本队列接纳条目的顺序,不足以替跨进程业务建立全局时间线。

还可把同一个任务对象连续入队两次,确认它会成为两个独立条目;优先队列不会自动识别重复请求,去重规则需要另行约定。

参考资料

文章版权声明:除非注明,否则均为云鹊BLOG原创文章,转载或复制请以超链接形式并注明出处。