Python如何实现轻量级的异步任务队列_仅使用asyncio实现内存队列-Python教程-PHP中文网

asyncio.Queue 是原生协程安全的轻量内存队列，需在同事件循环中使用，支持背压但无持久化；正确用法是 producer/consumer 模式配合 task_done() 和 join()，须设 maxsize 防 OOM。

python如何实现轻量级的异步任务队列_仅使用asyncio实现内存队列

asyncio.Queue 就是你要的轻量队列

不用装第三方包，asyncio.Queue 原生支持协程间安全的数据传递，内存驻留、无序列化开销、启动零成本。它不是“模拟”队列，而是 asyncio 事件循环原生调度的同步原语。

常见错误是拿它当线程安全队列用——比如在 threading.Thread 里调用 await queue.put()，这会直接报 RuntimeError: no running event loop。它只在协程上下文里有效。

必须在同一个 asyncio.EventLoop 中使用（通常就是主协程或同个 asyncio.create_task() 启动的任务）
maxsize=0（默认）表示无限制；设为正整数后，put() 在满时会 await，自动实现背压
不支持持久化、不跨进程、不跨解释器——就是纯内存+协程协作

如何启动一个持续消费任务的 worker

别写死循环加 sleep()，那会阻塞事件循环。正确做法是用 async for 或持续 await queue.get()，让出控制权给其他协程。

典型结构是：一个 producer 协程不断 await queue.put(item)，一个或多个 consumer 协程用 while True: + await queue.get() 拉取并处理。

记得调用 queue.task_done()，否则 queue.join() 永远不会返回
consumer 内部异常要捕获，否则任务崩溃后整个 worker 就静默退出了
如果想优雅停止，可以用 asyncio.Event 配合 queue.get_nowait() 或超时 await asyncio.wait_for(queue.get(), timeout=1)

async def worker(queue: asyncio.Queue):
    while True:
        try:
            item = await queue.get()
            await process(item)  # 实际异步处理
        except Exception as e:
            print(f"failed on {item}: {e}")
        finally:
            queue.task_done()