设置maxsize时，asyncio.Queue异步put的优势及入队方案合理性疑问

我有一个同步调用函数，负责将任务放入asyncio.Queue，以供异步worker处理。当前代码如下：

import asyncio
import os
import sys

queue = asyncio.Queue(maxsize=0)

async def worker():
    try:
        while True:
            task = await queue.get()
            print(f'processing task: {task}')
            queue.task_done()
    except asyncio.QueueShutDown:
        return

def on_read():
    data = os.read(0, 10)
    if data:
        queue.put_nowait(data)
    else:
        queue.shutdown()

loop = asyncio.new_event_loop()
loop.add_reader(0, on_read)
loop.run_until_complete(worker())

该代码可通过以下方式运行：

for i in $(seq 3); do echo $i; sleep 1; done | python3 minimal_queue.py

目前我使用put_nowait向队列中添加任务，理论上这可能触发QueueFull异常，但由于队列是无界的（maxsize=0），这种情况不会发生。不过这可能导致队列持续增长直至系统内存耗尽。

我曾考虑三种方案：

• 保留无界队列，继续用put_nowait：风险是内存耗尽
• 为队列设置maxsize，然后使用loop.create_task(queue.put(data))：担心高负载下堆积大量阻塞任务，更快耗尽内存
• 保留put_nowait()并设置maxsize：更早触发失败，但不会耗尽系统所有内存，对其他进程影响更小

我目前倾向于第三种方案，但不确定自己是否完全理解这些队列的工作机制，我的推理是否正确？

你的思路是正确的，第三种方案确实是更稳妥的选择，下面逐一分析三种方案的特性：

方案1：无界队列 + put_nowait #

无界队列的maxsize=0意味着可以无限接收任务，put_nowait永远不会抛出QueueFull。但当生产者速度远快于worker处理速度时，队列会持续膨胀，直到耗尽系统可用内存，严重时会导致进程被OOM killer终止，甚至影响整个系统稳定性。这种方案仅适用于生产者与处理速度完全匹配的场景，否则风险极高。

方案2：有界队列 + loop.create_task(queue.put(data)) #

用loop.create_task将queue.put包装成异步任务，本质是让“等待队列空闲”的逻辑在异步后台执行，但存在两个关键问题：

• 每调用一次就创建一个新异步任务，队列满时这些等待中的任务会堆积在事件循环的任务列表里，每个任务都占用内存。若生产者持续高速输入，任务列表的内存消耗可能比队列本身膨胀得更快，反而加速内存耗尽。
• 同步函数on_read由事件循环的add_reader触发，本身是同步执行的，堆积的任务会占用事件循环调度资源，可能影响worker的执行效率。

方案3：有界队列 + put_nowait #

设置合理的maxsize后，队列满时put_nowait会立即抛出QueueFull异常，优势很明显：

• 能严格限制队列的内存占用，避免无限制膨胀，保护系统内存资源，对其他进程影响最小。
• 异常触发及时，你可以在on_read中捕获异常，做针对性处理（比如暂时丢弃任务、记录日志、通知生产者放缓速度等）。

需要注意，使用这个方案时必须捕获QueueFull异常，否则未处理的异常会导致事件循环终止。修改后的on_read示例：

def on_read():
    data = os.read(0, 10)
    if data:
        try:
            queue.put_nowait(data)
        except asyncio.QueueFull:
            # 自定义处理逻辑，比如记录错误日志
            print("队列已满，暂时无法添加任务", file=sys.stderr)
    else:
        queue.shutdown()

另外，maxsize的取值需要根据业务场景调整：若worker处理速度稳定，可设置为worker数量的2-3倍；若处理波动大，可适当增大，但不要设置过大失去限制意义。

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