Python3.9 asyncio学习：队列依赖任务异步实现及QueueEmpty报错排查

问题2解答：QueueEmpty报错原因 #

• 核心原因是所有工作协程都使用了get_nowait()方法读取队列：该方法不会等待队列生产数据，只要读取瞬间队列为空，就会直接抛出asyncio.queues.QueueEmpty异常。比如worker_2、worker_3启动时，上游的worker_1还没完成数据处理、还没往队列写入数据，这两个协程直接读空队列就会立刻报错。
• 其次asyncio.gather的调用存在语法错误：原代码写的是await asyncio.gather(*[tasks_1+tasks_2+tasks_3])，多套了一层列表，相当于把整个任务列表作为第一个参数传给gather，无法正常调度协程，也会加重运行异常。
• 额外的错误逻辑：worker_1、worker_2里写了await queue_2.join()、await queue_3.join()，每次写入一条数据就等待下游队列所有任务处理完成，会阻塞上游协程运行，导致下游队列长期为空，进一步提升了读空报错的概率。

问题1解答：代码改进方案 #

核心修改点如下：

1. 把所有get_nowait()替换为await queue.get()：协程会等待队列有数据后再读取，不会读空报错
2. 修复asyncio.gather的调用错误，正确解包所有任务
3. 删除工作协程中不必要的await queue.join()逻辑，避免阻塞并发
4. 调整退出逻辑：主函数先等待所有上游队列的任务全部处理完成，再主动取消所有工作协程，避免用不可靠的queue.empty()作为退出判断
5. 正确配置队列的task_done()调用，保证队列join()逻辑正常工作

改进后的可运行代码：

#!/usr/bin/env python 3.9
# -*- coding: utf-8 -*-
import asyncio

def worker_0(data:list, queue_1:asyncio.Queue):
    for dat in data:
        k,v = dat['key'], dat['value']
        queue_1.put_nowait({"key":k, "value":"q1_"+str(v)})
        print(f"work_0 done!  Production:  'key':{k}, 'value':{'q1_'+str(v)}     \tq1_size: {queue_1.qsize()}")

async def worker_1(queue_1:asyncio.Queue, queue_2:asyncio.Queue):
    while True:
        data = await queue_1.get()
        k,v = data['key'], data['value']
        await asyncio.sleep(1.5)
        queue_2.put_nowait({"key":k, "value":"q2_"+str(v)})
        queue_1.task_done()
        print(f"work_1 done!  Consumption：'key': {k}, 'value': {v}               \tq1_size: {queue_1.qsize()}")
        print(f"work_1 done!  Production： 'key': {k}, 'value': {'q2_' + str(v)}  \tq2_size: {queue_2.qsize()}")

async def worker_2(queue_2:asyncio.Queue, queue_3:asyncio.Queue):
    while True:
        data = await queue_2.get()
        k, v = data['key'], data['value']
        await asyncio.sleep(2)
        queue_3.put_nowait({"key": k, "value": "q3_" + str(v)})
        queue_2.task_done()
        print(f"work_2 done!  Consumption：'key': {k}, 'value':  {v}              \tq2_size: {queue_2.qsize()}")
        print(f"work_2 done!  Production： 'key': {k}, 'value':  {'q3_' + str(v)} \tq3_size: {queue_3.qsize()}")

async def worker_3(queue_3:asyncio.Queue):
    while True:
        data = await queue_3.get()
        k, v = data['key'], data['value']
        await asyncio.sleep(2.5)
        result = str(k) + str(v)
        print(f"result:{result}")
        print(f"work_3 done!  Consumption：'key': {k}, 'value': {v}               \tq3_size: {queue_3.qsize()}")
        queue_3.task_done()

async def main():
    queue_1 = asyncio.Queue()
    queue_2 = asyncio.Queue()
    queue_3 = asyncio.Queue()
    # 先生产所有初始数据
    worker_0(data, queue_1)
    # 创建所有工作任务
    tasks_1 = [asyncio.create_task(worker_1(queue_1, queue_2)) for _ in range(10)]
    tasks_2 = [asyncio.create_task(worker_2(queue_2, queue_3)) for _ in range(10)]
    tasks_3 = [asyncio.create_task(worker_3(queue_3)) for _ in range(10)]
    all_tasks = tasks_1 + tasks_2 + tasks_3

    # 等待所有队列的任务全部处理完成
    await queue_1.join()
    await queue_2.join()
    await queue_3.join()

    # 所有任务处理完成后，取消所有工作协程
    for task in all_tasks:
        task.cancel()
    # 等待协程退出完成，忽略取消异常
    await asyncio.gather(*all_tasks, return_exceptions=True)

if __name__ == '__main__':
    data = [{"key":i,"value":v} for i,v in enumerate(range(100)) ]
    asyncio.run(main())

内容的提问来源于stack exchange，提问作者CG Zhang