为何asyncio.timeout包装异步生成器时无法生效？

先看你给出的第一个示例代码：

async def generate_many_numbers():
    for i in range(1000000000):
        yield i

async def main2():
    async with asyncio.timeout(1):
        async for number in generate_many_numbers():
            print(number)

asyncio.run(main2())

这个超时不生效的核心原因是异步生成器里没有任何await操作：

• Asyncio的事件循环是协作式调度的，只有当任务主动通过await让出CPU时，事件循环才有机会去检查超时、任务取消等调度信号。
• 你的generate_many_numbers里只有同步的for循环和yield，全程霸占着CPU执行同步代码，事件循环根本插不上手，自然无法触发超时逻辑。

这和PEP 789描述的问题一致吗？ #

是的，完全一致。PEP 789正是针对异步生成器迭代时无法被中断、超时无法触发的问题提出的解决方案——本质就是异步生成器的__anext__方法如果没有await点，会一直阻塞事件循环，导致外部的超时、取消信号无法被处理。

为什么添加await asyncio.sleep(0)能解决问题？ #

asyncio.sleep(0)是一个特殊的await调用，它不会让任务真的休眠，而是主动触发一次事件循环的调度，让当前任务暂时让出CPU。这样每次生成器yield之后，事件循环都有机会检查超时是否已经到达，从而正确触发TimeoutError。

修改后的代码示例：

async def generate_many_numbers():
    for i in range(1000000000):
        await asyncio.sleep(0)
        yield i

async def iterate_through_numbers():
    async for number in generate_many_numbers():
        print(number)

async def main2():
    async with asyncio.timeout(1):
        async for number in generate_many_numbers():
            print(number)

async def main3():
    try:
        await asyncio.wait_for(iterate_through_numbers(), timeout=1)
    except asyncio.TimeoutError:
        print("The task timed out!")

这里不管是用asyncio.timeout包装async for，还是用asyncio.wait_for包装整个迭代任务，都能正常触发超时——因为生成器每次迭代都会通过sleep(0)让出CPU，给事件循环处理超时的机会。

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