如何在Python中实现一个支持并发操作的优先级阻塞队列？-Python教程-PHP中文网

queue.PriorityQueue不支持async/await，纯异步场景下无法直接await get()/put()；需改用Python 3.12+的asyncio.PriorityQueue。

如何在python中实现一个支持并发操作的优先级阻塞队列？

为什么不能直接用 `queue.PriorityQueue` 做并发任务调度？

queue.PriorityQueue 本身是线程安全的，但它不支持协程（async/await），在 asyncio 场景下直接 await 它会报 TypeError: object PriorityQueue can't be used in 'await' expression。如果你混合使用多线程和 asyncio（比如主线程跑 event loop，又启了 ThreadPoolExecutor 来调用阻塞 IO），那它能用；但纯异步服务里，它就是个“哑巴队列”——没法 await get()，也没法 await put()。

常见错误现象：写 await pq.get() 直接崩溃；或退而求其次用 loop.run_in_executor 包一层，结果引入不必要的线程切换开销，还容易卡死（比如 executor 队列满、或 get() 在 executor 里阻塞住整个线程池）。

实操建议：

纯 asyncio 项目 → 必须换用 asyncio.PriorityQueue（Python 3.12+ 原生支持）
Python asyncio.Condition + heapq
需要同时支持线程和协程 → 不推荐，应分层：上层用 async 队列，底层 IO 操作再丢给线程池

Python 3.12+ 怎么直接用 `asyncio.PriorityQueue`？

3.12 真正把 asyncio.PriorityQueue 加进标准库了，接口和 queue.PriorityQueue 高度一致，但所有方法都返回可 await 对象。

注意点：

构造时不传 maxsize 默认无上限；传了之后 put() 会在满时自动 await，直到有空位
元素必须是可比较的元组，比如 (priority, item)；如果 item 不可比（如 dict），会抛 TypeError
get() 返回的是元素本身（不是元组），即你 put((1, {'id': 42}))，get() 拿到的是 {'id': 42}

示例：

import asyncio
<p>async def main():
q = asyncio.PriorityQueue()
await q.put((2, "low-pri"))
await q.put((1, "high-pri"))
print(await q.get())  # 输出: "high-pri"
print(await q.get())  # 输出: "low-pri"</p><p>asyncio.run(main())
</p>

Python

关键不是“堆排序”，而是“如何让多个协程公平等待并响应优先级变化”。直接用 heapq 管理内部列表没问题，但 get() 必须配合 asyncio.Condition 实现“谁优先级高谁先被唤醒”，而不是 FIFO 唤醒。

Python Hardener

通过完整的安全审查、代码整洁重构、健壮的错误处理、正确的日志记录以及完整的文档字符串覆盖，强化 Python 脚本。

下载

容易踩的坑：

只用 asyncio.Event 或简单 asyncio.Lock → 唤醒顺序不可控，低优先级任务可能先抢到锁
每次 put() 都 notify_all() → 浪费 CPU，尤其队列大时大量协程被唤醒又立刻 continue 判断
没处理 task.cancelled() → 被取消的等待协程可能还占着 condition 的 wait list，导致后续 notify 失效

实操建议：内部用 heapq 存元组 (priority, insertion_order, item)，其中 insertion_order 是单调递增计数器，解决相同优先级下的插入顺序问题；get() 中用 condition.wait_for(lambda: q) 检查非空，且每次 notify 只唤醒一个协程（notify(1)）。