asyncio.Queue 不能直接支持优先级,因其底层为 FIFO 队列,无内置优先级比较逻辑;需结合 heapq 与 asyncio.Lock 实现线程安全的异步优先级队列,元素推荐用 (priority, insertion_count, item) 元组确保可比性。

为什么 asyncio.Queue 不能直接支持优先级
因为 asyncio.Queue 底层是 FIFO(先进先出)队列,没有内置优先级比较逻辑。你调用 put() 或 get() 时,它不会按数值大小或自定义规则排序——哪怕你传入的是元组 (priority, item),它也只当普通对象存取,不触发比较。
用 heapq + asyncio.Lock 实现线程安全的异步优先级队列
标准库 heapq 支持堆操作,但本身不是线程/协程安全的;而 asyncio.Queue 虽安全却无优先级。折中方案是:用 heapq 管理内部列表,用 asyncio.Lock 保护所有读写操作。
关键点:
- 元素必须可比较:推荐用
(priority, insertion_count, item)元组,避免item自身不可比导致heapq报TypeError: ' -
insertion_count是递增整数,确保相同优先级时按插入顺序出队(稳定排序) - 所有
put/get操作必须在async with lock:下执行,否则并发put可能破坏堆结构 - 不要直接暴露内部
self._heap,否则外部修改会绕过锁和堆维护逻辑
示例核心片段:
立即学习“Python免费学习笔记(深入)”;
import heapq
import asyncio
class AsyncPriorityQueue:
def __init__(self):
self._heap = []
self._lock = asyncio.Lock()
self._counter = 0
async def put(self, item, priority=0):
async with self._lock:
heapq.heappush(self._heap, (priority, self._counter, item))
self._counter += 1
async def get(self):
async with self._lock:
if not self._heap:
raise IndexError("pop from empty priority queue")
return heapq.heappop(self._heap)[2]
asyncio.PriorityQueue 在 Python 3.12+ 中终于原生可用
Python 3.12 引入了 asyncio.PriorityQueue,行为与 queue.PriorityQueue 类似,但基于协程。它内部已处理锁和堆逻辑,开箱即用。
注意:
- 仅限 Python ≥ 3.12,低版本会报
AttributeError: module 'asyncio' has no attribute 'PriorityQueue' - 元素仍需可比较;若传入
(priority, item),要求item也支持<,否则可能崩溃 —— 建议统一用(priority, counter, item)避坑 - 它不兼容
asyncio.Queue的全部方法(如无task_done()),若需任务追踪得自行封装
实际调度中容易忽略的阻塞点
很多人以为“放进优先级队列就自动按序执行”,其实不然。队列只管顺序出队,真正调度还得靠消费者协程主动 await queue.get() 并处理。常见疏漏:
- 多个消费者协程同时
await queue.get(),高优任务未必被高优协程拿到——优先级只作用于队列内部顺序,不绑定执行者 - 任务处理耗时长时,后续
get()会被阻塞,即使有更高优任务入队也无法抢占;真正的“抢占式调度”需配合取消机制(asyncio.cancel()+try/except CancelledError) - 未设置
maxsize时,大量低优任务涌入可能把高优任务“压”在底部,内存持续增长
优先级队列解决的是“谁先拿到任务”,不是“谁先完成任务”。真要控制执行节奏,得结合限流、超时、取消等手段,不能只依赖队列本身。


















