Python协程队列详解（异步编程中的高效任务调度利器）

什么是协程队列？

asyncio.Queue 是 Python 标准库 asyncio 模块提供的一个线程安全（在单线程异步上下文中）的队列结构，专为协程设计。它允许你在多个协程之间安全地传递数据，常用于生产者-消费者模型。

与普通队列不同，asyncio.Queue 的操作（如 put() 和 get()）都是 异步方法，这意味着它们不会阻塞整个事件循环，而是挂起当前协程，让出控制权给其他任务，从而实现高并发。

基本用法：创建和操作队列

首先，我们来看一个最简单的例子：创建一个队列，放入数据，再取出数据。

import asyncioasync def main():    # 创建一个最大容量为3的队列    queue = asyncio.Queue(maxsize=3)        # 异步放入数据    await queue.put("任务1")    await queue.put("任务2")        # 异步取出数据    item1 = await queue.get()    print(f"取出: {item1}")        item2 = await queue.get()    print(f"取出: {item2}")# 运行协程asyncio.run(main())

运行结果：

取出: 任务1取出: 任务2

实战案例：生产者-消费者模型

这是协程队列最常见的应用场景。多个“生产者”协程生成任务并放入队列，多个“消费者”协程从队列中取出任务并处理。

import asyncioimport random# 生产者协程async def producer(queue, name):    for i in range(5):        item = f"{name}-任务{i}"        await queue.put(item)        print(f"[生产者 {name}] 放入: {item}")        await asyncio.sleep(random.uniform(0.1, 0.5))  # 模拟耗时# 消费者协程async def consumer(queue, name):    while True:        item = await queue.get()        if item is None:  # 退出信号            break        print(f"[消费者 {name}] 处理: {item}")        await asyncio.sleep(random.uniform(0.2, 0.6))  # 模拟处理时间        queue.task_done()  # 告知队列任务已完成async def main():    queue = asyncio.Queue()    # 启动2个生产者和3个消费者    producers = [        asyncio.create_task(producer(queue, "P1")),        asyncio.create_task(producer(queue, "P2"))    ]        consumers = [        asyncio.create_task(consumer(queue, "C1")),        asyncio.create_task(consumer(queue, "C2")),        asyncio.create_task(consumer(queue, "C3"))    ]    # 等待所有生产者完成    await asyncio.gather(*producers)        # 等待队列中所有任务被消费完毕    await queue.join()        # 发送退出信号给消费者    for _ in consumers:        await queue.put(None)        # 等待消费者结束    await asyncio.gather(*consumers)# 运行主协程asyncio.run(main())

在这个例子中，我们使用了 queue.task_done() 和 queue.join() 来确保所有任务都被正确处理。这是构建健壮异步系统的关键技巧。

常用方法总结

• queue.put(item)：异步放入一个项目（如果队列满则等待）
• queue.get()：异步取出一个项目（如果队列空则等待）
• queue.qsize()：返回当前队列大小（注意：在多协程环境下可能不准确）
• queue.empty() / queue.full()：判断队列是否为空或满
• queue.task_done()：标记一个任务已完成（必须与 get() 配对使用）
• queue.join()：阻塞直到队列中所有任务都调用了 task_done()

为什么使用协程队列？

使用 Python协程队列 可以带来以下优势：

• ✅ 高效并发：避免线程切换开销，单线程内实现高并发
• ✅ 资源节约：相比多线程/多进程，内存占用更低
• ✅ 逻辑清晰：生产者-消费者模型解耦任务生成与处理
• ✅ 可控性强：通过队列大小限制防止内存爆炸

结语

掌握 asyncio队列 是进阶 异步编程Python 的关键一步。无论你是开发 Web 爬虫、API 服务，还是处理大量 I/O 密集型任务，协程任务调度 都能显著提升你的程序性能和响应速度。

现在就动手试试吧！你可以修改上面的代码，调整生产者/消费者的数量，观察输出结果的变化，加深对协程队列工作原理的理解。

关键词回顾：Python协程队列、asyncio队列、异步编程Python、协程任务调度