asyncio.Queue不支持优先级，需用heapq+asyncio.Lock自建优先级队列；优先级数字越小越先执行，需加id(item)避免不可比较对象报错，用asyncio.Event减少忙等开销。
asyncio 里没有内置优先级队列，asyncio.Queue 是 FIFO
这是最常被误以为“有”的地方——asyncio.Queue 确实不支持优先级。它底层用的是 collections.deque，只保证先进先出。一旦你往里面 put() 三个任务，哪怕第三个是高优紧急任务，它也得等前两个 get() 完才能轮到。
真实场景中，比如：一个服务同时处理用户普通请求（低优）、支付回调（中优）、风控拦截信号（高优），靠默认 asyncio.Queue 会卡住关键路径。
- 别试图给
asyncio.Queue加priority参数——它根本没有这个参数 - 别在
put()时手动排序——并发写入下竞态条件极难控制 - 真正可用的解法是换底层数据结构：用
heapq+asyncio.Lock自建
用 heapq + asyncio.Lock 实现线程安全的异步优先级队列
Python 的 heapq 本身不是线程安全的，但在 asyncio 单线程事件循环里，只要确保所有 put/get 都走同一个协程入口，并加锁，就能避免堆损坏。重点不是“多线程安全”，而是“协程间操作原子性”。
示例核心逻辑：
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
import heapq
import asyncio
<p>class PriorityQueue:
def <strong>init</strong>(self):
self._queue = []
self._lock = asyncio.Lock()
self._event = asyncio.Event()</p><pre class="brush:python;toolbar:false;">async def put(self, item, priority=0):
async with self._lock:
heapq.heappush(self._queue, (priority, id(item), item))
self._event.set() # 唤醒等待者
async def get(self):
while True:
async with self._lock:
if self._queue:
_, _, item = heapq.heappop(self._queue)
if not self._queue:
self._event.clear()
return item
await self._event.wait()
-
id(item)是为避免不可比较对象（如 dict）触发TypeError - 优先级数字越小越先执行，和 Unix 进程优先级一致，别反着设
-
self._event替代忙等，降低空转开销；但要注意：多个get协程唤醒后仍需重新抢锁判断队列是否为空
优先级值怎么设才不容易翻车
很多人随手用字符串或枚举当优先级，结果 heapq 比较时报 TypeError: '。必须确保元组第一个元素可比较、且语义清晰。
- 推荐用整数：
0（最高优）、10（默认）、100（后台批处理）——留足扩展余地 - 避免用时间戳做优先级（如
time.time()），浮点精度+时钟漂移会导致顺序错乱 - 不要混用类型：比如有时传
"high"，有时传0，heapq会炸 - 如果业务需要动态调整某任务优先级，不能改已入堆的元组——得重投新任务，旧任务在
get时跳过（需额外标记字段）
和 asyncio.create_task() 配合时的关键陷阱
直接把 PriorityQueue.get() 放进 create_task() 循环里，看似合理，实则埋雷：一旦某个任务抛异常没捕获，整个消费者协程就退出，后续任务全卡住。
- 每个消费者协程必须包
try/except，且至少 log 错误，否则静默失败 - 别让单个消费者承担全部压力——按优先级分多个消费者：高优单独一个
while True循环，低优另起一个，避免低优长任务拖垮高优吞吐 -
asyncio.wait_for()要慎用在get()上：超时后协程退出，但锁可能还占着，导致死锁；建议用asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED)配合self._event
优先级不是加个数字就完事；队列结构、锁粒度、错误传播、消费者拓扑，每一层都可能让“关键任务”变“失踪任务”。最常漏掉的是消费者异常退出后无人重启，以及高优任务被低优任务的 await 卡住——后者得靠分队列或取消机制兜底。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
