queue.PriorityQueue不支持async/await，纯异步场景下无法直接await get()/put()；需改用Python 3.12+的asyncio.PriorityQueue。
为什么不能直接用 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 对象。
立即学习“Python免费学习笔记（深入）”；
注意点：
- 构造时不传
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())
Python
关键不是“堆排序”，而是“如何让多个协程公平等待并响应优先级变化”。直接用 heapq 管理内部列表没问题，但 get() 必须配合 asyncio.Condition 实现“谁优先级高谁先被唤醒”，而不是 FIFO 唤醒。
容易踩的坑：
- 只用
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)）。
并发场景下，优先级和公平性怎么权衡？
真实服务中，“高优任务插队”和“饿死低优任务”往往是一体两面。比如日志上报设为最低优先级，但如果持续有高优请求进来，它可能永远得不到执行。
解决方案不是加“时间戳老化”，而是控制队列行为边界：
- 设置
maxsize强制背压：避免无限堆积，让生产者主动降速或失败 - 给低优任务加“最大等待时间”：用
asyncio.wait_for(q.get(), timeout=30)，超时后降级重试或丢弃 - 不用全局单队列：按业务域拆成多个队列（如
user_queue,admin_queue），再用外部调度器做跨队列优先级仲裁
最常被忽略的一点：优先级数字越小越高，但很多人误用 (time.time(), item) 当优先级——浮点时间戳精度不够，同一毫秒内插入的多个任务比较会出错，必须用 (time.time(), counter, item) 或直接用 itertools.count()。
