asyncio.priorityqueue是协程安全的优先级队列，put()和get()需显式await；存入的是未执行的协程对象，取出后必须await才能运行，否则触发runtimewarning或静默失败。

如何在python 3.11中使用asyncio.priorityqueue处理并发任务

asyncio.PriorityQueue 是协程安全的，但不自动 await 任务

它只是带优先级的线程/协程安全队列，put() 和 get() 是异步方法，必须显式 await。直接 queue.put(task) 不会执行任务，也不会报错——只是把协程对象塞进队列里，后续取出来如果不 await，就等于白放。

常见错误现象：RuntimeWarning: coroutine 'my_worker' was never awaited，或程序静默退出、任务全没运行。

使用场景：需要按优先级调度多个异步 I/O 任务（如爬虫请求、日志上传、消息重试）
put() 接收任意可比较对象作为优先级，通常用 (priority, item) 元组
注意：优先级值越小越先被 get() 取出；若优先级相同，按插入顺序（FIFO）

正确启动任务需在 consumer 中 await 协程对象

从 asyncio.PriorityQueue 取出的内容，如果是协程对象（比如 async def 函数返回值），必须 await 才真正执行。不能当成普通函数调用。

import asyncio
<p>async def fetch_url(url, priority):
await asyncio.sleep(0.1)  # 模拟网络延迟
return f"done {url}"</p><p>async def worker(queue):
while True:
try:</p><h1>注意：这里取出的是 (priority, coro)，不是结果</h1><pre class="brush:python;toolbar:false;">        priority, coro = await queue.get()
        result = await coro  # ✅ 必须 await 这个协程
        print(result)
    except asyncio.CancelledError:
        break

async def main(): queue = asyncio.PriorityQueue()

✅ 正确：传入已构造好的协程对象

await queue.put((10, fetch_url("low-prio.com", 10)))
await queue.put((1, fetch_url("high-prio.com", 1)))

task = asyncio.create_task(worker(queue))
await asyncio.sleep(0.2)
task.cancel()

避免在 put 时提前 await 导致失去优先级控制

如果写成 await queue.put((p, await some_async_func()))，等于先执行完任务再塞进队列，优先级就只控制“存放顺序”，完全失去调度意义。

python-script-generator

快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构，支持CLI、API、爬虫、Bot、Django等多种项目类型，包含完整的项目结构、配置文件、依赖管理、测试、README和文档。

下载

错误示范：await queue.put((5, await fetch_url("a.com"))) → 阻塞等待完成，再入队
正确做法：入队的是未执行的协程对象，由 consumer 统一 await，确保优先级生效
若需预计算参数（如 URL、token），可在入队前同步算好，但不要 await 异步操作

并发 consumer 数量影响实际吞吐，但不改变优先级语义

启多个 worker 任务并行消费，能提升整体处理速度，但 asyncio.PriorityQueue 本身不保证跨 worker 的全局优先级抢占——因为 get() 是协程级别的原子操作，一旦某个 worker 成功 await get() 返回，该任务就归属它了。

也就是说：高优任务不会“插队”正在执行中的低优任务，但会在所有 worker 的下一次 get() 调用中被优先选中。

性能影响：consumer 太少 → 高优任务可能排队等待低优任务结束；太多 → 增加调度开销，且无法突破单次 get() 的原子性限制
兼容性：Python 3.11 中 asyncio.PriorityQueue 行为与 3.9+ 一致，无新增 breaking change
容易忽略的点：没有内置“重入队”或“延迟重试”机制，需自己封装逻辑（比如捕获异常后 await queue.put((new_priority, coro))）

实际用起来，最常卡住的地方不是语法，而是混淆了“协程对象”和“协程执行结果”。只要记住：队列存的是“待办事项”，不是“已完成报告”。