Python asyncio.PriorityQueue 怎样保持相同优先级的稳定顺序

我在把几个异步任务放进优先级队列时，最先遇到的不是“优先级怎么写”，而是同级任务为什么偶尔顺序不固定。asyncio.PriorityQueue 会按最低优先级先取出，官方文档给出的典型队列项是 (priority_number, data)；如果 data 是字典或自定义对象，比较走到第二项时还可能直接失败。

官方文档：https://docs.python.org/3/library/asyncio-queue.html

比较稳妥的写法是把队列项扩展成 (priority, sequence, payload)：第一项决定优先级，第二项固定同级任务的先后，第三项只保存真正要处理的数据。这样既能保持相同优先级的入队顺序，也不会要求字典或业务对象互相比较。

先看清 PriorityQueue 到底比较什么

asyncio.PriorityQueue 是 asyncio.Queue 的变体，取出顺序是最低值优先。它并不知道“业务上的紧急程度”，只会依赖队列项的可比较关系。因此下面这个结构表达的是“数值越小越优先”：(10, "普通任务") 会排在 (1, "紧急任务") 后面。

真正容易踩坑的是同级项。若两个元素分别是 (1, {"id": 1}) 和 (1, {"id": 2})，比较器已经无法只靠第一项分出先后，继续比较字典就会把排序细节暴露给业务数据。解决思路不是给 payload 强行实现排序，而是为每次入队分配一个独立序号。

用 sequence 固定同级任务的先后

把序号放在 payload 前面即可形成稳定的排序键。序号可以用 itertools.count() 产生，它只负责给本次进程里的入队动作编号，不需要把编号写入业务对象。

import asyncio
import itertools

# sequence 只表示入队先后，不改变任务自身的数据结构。
sequence = itertools.count()
queue = asyncio.PriorityQueue()

async def enqueue(priority, payload):
    # 数值越小越先取出；sequence 让相同 priority 仍有确定顺序。
    item = (priority, next(sequence), payload)
    await queue.put(item)

async def main():
    # 三个任务优先级相同，取出顺序由序号保持为 A、B、C。
    await enqueue(5, {"name": "A"})
    await enqueue(5, {"name": "B"})
    await enqueue(5, {"name": "C"})

    for _ in range(3):
        priority, order, payload = await queue.get()
        try:
            # 业务代码只消费 payload，排序字段不泄漏到处理逻辑。
            print(priority, order, payload["name"])
        finally:
            # 每次 get 都要对应一次 task_done，join 才能正确结束。
            queue.task_done()

    # 等待所有已取出的项目都被标记为完成。
    await queue.join()

asyncio.run(main())

这段结构的关键是 tuple 的前两项总能比较：优先级不同先按 priority 排，同级时按 sequence 排。只有前两项都相等才会触及 payload，而单调递增序号使正常入队路径不会走到这一步。

为什么不要让字典或对象承担排序

业务 payload 往往是字典、dataclass 或第三方对象，它们的字段变化不应该悄悄改变调度顺序。把 payload 放到第三项有两个好处：一是排序契约集中在队列封装中，二是消费者不必知道队列内部为何能稳定出队。

如果业务确实需要同级任务按某个字段再排序，可以明确把那个字段放在 sequence 的位置，并保证它的类型稳定；不要依赖字典插入顺序或对象的默认比较行为。

把生产者和消费者的职责封装起来

实际项目里通常有多个生产者和一个或多个消费者。建议把“生成排序键”放在生产者入口，把 task_done() 放在消费者的 finally 中，避免异常时遗漏未完成任务计数。

import asyncio
import itertools

class StablePriorityQueue:
    def __init__(self, maxsize=0):
        # maxsize 为正数时，put 会在容量用尽后等待空位。
        self._queue = asyncio.PriorityQueue(maxsize=maxsize)
        self._sequence = itertools.count()

    async def put(self, priority, payload):
        # 只在这一层组合排序键，调用方不需要手动维护序号。
        await self._queue.put((priority, next(self._sequence), payload))

    async def get(self):
        # 只返回业务需要的三项，消费者仍可在需要时读取 priority 和 order。
        return await self._queue.get()

    def task_done(self):
        # 委托给内部队列，保持 join 的未完成任务计数一致。
        self._queue.task_done()

    async def join(self):
        # 等待所有 put 进来的项目都完成 task_done。
        await self._queue.join()

async def worker(name, work_queue):
    while True:
        priority, order, payload = await work_queue.get()
        try:
            # 这里替换成真实 I/O 或业务处理；异常仍会触发 finally。
            await asyncio.sleep(0)
            print(name, priority, order, payload)
        finally:
            work_queue.task_done()

这里没有把消费者写成“处理完就自动结束”。join() 只关心每个入队项目是否最终调用了 task_done()，消费者生命周期仍应由项目自己的取消、关闭或哨兵策略管理。当前文章只讨论排序键和完成计数，不把队列关闭协议混进排序规则。

四个容易被忽略的边界

1. 优先级方向：官方语义是最低值优先。若业务认为数字越大越紧急，可以存入负值，或者在封装层统一转换，不能让不同生产者各自理解。
2. 序号的作用域：itertools.count() 只保证当前进程内由同一个计数器产生的顺序。跨进程、跨机器或需要持久化恢复时，应把排序键设计成业务可恢复的字段。
3. 取消与异常：消费者拿到项目后即使处理失败，也要在 finally 中调用 task_done()；否则等待 join() 的协程会一直保留未完成计数。
4. 容量限制：maxsize 控制的是队列容器容量，不是同一优先级的数量。容量为正时，生产者可能在 put() 处等待，设计时要把这个背压点纳入任务调度。

什么时候该选这个写法

只要任务有“优先级 + 同级按到达顺序处理”的要求，而且 payload 不天然支持比较，就应该使用三元组方案。它的代价很小：增加一个整数序号，并让队列封装多做一次拆包。

如果同级任务本来就允许任意顺序，或者 payload 是简单数字并且需要按 payload 二次排序，二元组也可以。但这个选择应写在队列接口约定里，避免后来有人把字典或对象塞进相同优先级的二元组后才发现比较异常。

常见问题

PriorityQueue 是数字越大越优先吗？

不是，默认是最低值优先。需要“大数更紧急”时，把业务值映射成负数或在封装层转换。

为什么还要调用 task_done？

每次 get() 取出的项目都应对应一次 task_done()。join() 依靠这个未完成任务计数判断是否可以继续。

能否用时间戳代替 sequence？

不建议把普通时间戳当唯一序号：精度、时钟调整和并发下的相等都可能带来歧义。进程内稳定排序用单调计数器更直接；需要跨进程时再设计明确的可持久化排序键。