Python异步for循环执行困惑：为何代码按串行方式运行？

问题描述 #

我是asyncio新手，正在学习基础异步for循环的行为。编写的代码如下，原本预期在sleep期间可以获取循环中的下一个元素并开始处理，但实际代码却按串行执行，如同非异步程序一样伴随sleep停顿。例如，当代码执行“doing something else with 1”时，本应可以获取下一个元素并启动处理，同时等待第一个元素的sleep结束，但运行结果却为串行执行。请问我忽略了什么要点？

代码示例 #

import asyncio


class CustomIterator():
    def __init__(self):
        self.counter = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.counter >= 3:
            raise StopAsyncIteration
        await asyncio.sleep(1)
        self.counter += 1
        return self.counter


async def f(item):
    print(f"doing something with {item}")
    await asyncio.sleep(3)


async def f2(item):
    print(f"doing something else with {item}")
    await asyncio.sleep(2)


async def do_async_stuff():
    async for item in CustomIterator():
        print(f"got {item}")
        await f(item)
        await f2(item)


if __name__ == '__main__':
    asyncio.run(do_async_stuff())

运行输出 #

got 1
doing something with 1
doing something else with 1
got 2
doing something with 2
doing something else with 2
got 3
doing something with 3
doing something else with 3

问题原因 #

你忽略了异步代码并发执行的核心：需要主动将任务提交给事件循环，而非串行等待每个任务完成。具体来说：

1. async for 本身是串行迭代的：每次迭代必须等待当前__anext__方法执行完毕（包括里面的await asyncio.sleep(1)），才会获取下一个元素。
2. 循环内的await f(item)和await f2(item)也是串行等待：处理当前item的两个协程必须依次执行完，才会进入下一次async for迭代。

整个流程相当于：等待元素1生成 → 处理元素1的f → 处理元素1的f2 → 等待元素2生成 → 处理元素2的f → 处理元素2的f2... 完全没有并发空间。

解决方案 #

要实现并发处理多个元素，需要用asyncio.create_task()将每个元素的处理逻辑包装成异步任务，让事件循环在等待当前任务sleep时，去执行其他任务和迭代下一个元素。

修改后的代码如下：

import asyncio


class CustomIterator():
    def __init__(self):
        self.counter = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.counter >= 3:
            raise StopAsyncIteration
        await asyncio.sleep(1)
        self.counter += 1
        return self.counter


async def f(item):
    print(f"doing something with {item}")
    await asyncio.sleep(3)


async def f2(item):
    print(f"doing something else with {item}")
    await asyncio.sleep(2)

# 把单个元素的处理逻辑封装成独立协程
async def process_item(item):
    await f(item)
    await f2(item)


async def do_async_stuff():
    tasks = []
    async for item in CustomIterator():
        print(f"got {item}")
        # 创建任务，立即提交给事件循环，不等待执行完成
        task = asyncio.create_task(process_item(item))
        tasks.append(task)
    # 等待所有任务执行完毕
    await asyncio.gather(*tasks)


if __name__ == '__main__':
    asyncio.run(do_async_stuff())

修改后的运行效果（示例） #

got 1
got 2
got 3
doing something with 1
doing something with 2
doing something with 3
doing something else with 1
doing something else with 2
doing something else with 3

可以看到，元素1、2、3会依次生成（因为__anext__里的sleep是串行的，如果想让元素生成也并发，需要调整迭代器逻辑，但通常迭代器的元素生成是有依赖的），但每个元素的处理任务会并发执行，不再等待上一个元素处理完才开始下一个。

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