Python中`async for`比`for`更快的场景有哪些？

在Python中，可以用async for [...] in [异步生成器]替代for [...] in [普通生成器]。当生成器包含可转为协程的慢I/O操作时，这种替换能降低程序整体执行时间。下文会给出该场景的详细示例及不同场景下的执行时间解析表达式。

核心问题 #

这是async for替代for能提速的唯一场景吗？async for还有其他可实现提速的适用场景吗？若存在，请给出实用示例说明async for相比for的提速原理。

示例代码说明 #

• 请按照代码末尾的注释正确调用TEST_ASYNC_FOR()；
• 可自由修改BLK、NBLK、GEN和N参数进行测试。

async def TEST_ASYNC_FOR():
    
    import asyncio
    import time
    
    BLK = 0.15  # 在两个for循环中以`time.sleep(BLK)`形式出现
    NBLK = 0.16 # 在两个for循环中以`await asyncio.sleep(NBLK)`形式出现
    GEN = 0.17  # 在标准（异步）生成器中以`time.sleep(GEN)`（`await asyncio.sleep(GEN)`）形式出现
    N = 3       # 每个for循环的迭代次数
    
    def generator():
        for i in range(N):
            time.sleep(GEN) # 慢I/O操作会阻塞线程
            yield i
    
    async def async_generator():
        for i in range(N):
            await asyncio.sleep(GEN) # 慢I/O操作会将控制权交还给事件循环
            yield i
    
    async def standard_for():
        for i in generator():
            time.sleep(BLK) # CPU密集型操作会阻塞线程
            await asyncio.sleep(NBLK) # 慢I/O操作会将控制权交还给事件循环
    
    async def async_for():
        async for i in async_generator():
            time.sleep(BLK) # CPU密集型操作会阻塞线程
            await asyncio.sleep(NBLK) # 慢I/O操作会将控制权交还给事件循环
    
    t0 = time.perf_counter()
    await standard_for()
    print(f"1x for-loop:			    {time.perf_counter()-t0:.2f} | expected: {N*BLK + N*NBLK + N*GEN:.2f} = N*BLK + N*NBLK + N*GEN")
    
    t0 = time.perf_counter()
    await async_for()
    print(f"1x async for-loop:		    {time.perf_counter()-t0:.2f} | expected: {N*BLK + N*NBLK + N*GEN:.2f} = N*BLK + N*NBLK + N*GEN")
    
    def standard_for_duration(BLK, NBLK, GEN, N):
        t = 2*N*BLK + 2*N*GEN + NBLK + (N-1)*max(NBLK-(BLK+GEN), 0)
        return t, "2*N*BLK + 2*N*GEN + NBLK + (N-1)*max(NBLK-(BLK+GEN), 0)"
    
    t0 = time.perf_counter()
    await asyncio.gather(standard_for(), standard_for())
    t_st, expr_st = standard_for_duration(BLK, NBLK, GEN, N)
    print(f"2x for-loop (separate tasks):	    {time.perf_counter()-t0:.2f} | expected: {t_st:.2f} = {expr_st}")
    
    def async_for_duration(BLK, NBLK, GEN, N):
        t = 2*N*BLK + N*GEN + NBLK + (N-1)*max(NBLK-BLK, 0) + (N-1)*GEN*(BLK > NBLK)*(NBLK > GEN) + GEN*(NBLK > BLK)*(BLK > GEN)
        return t, "2*N*BLK + N*GEN + NBLK + (N-1)*max(NBLK-BLK, 0) + (N-1)*GEN*(BLK > NBLK)*(NBLK > GEN) + GEN*(NBLK > BLK)*(BLK > GEN)"
    
    t0 = time.perf_counter()
    await asyncio.gather(async_for(), async_for())
    t_async, expr_async = async_for_duration(BLK, NBLK, GEN, N)
    print(f"2x async for-loop (separate tasks): {time.perf_counter()-t0:.2f} | expected: {t_async:.2f} = {expr_async}")

await TEST_ASYNC_FOR() # 在Jupyter Lab中使用，环境已运行事件循环

# import asyncio
# asyncio.run(TEST_ASYNC_FOR()) # 在独立Python脚本中使用

代码输出 #

1x for-loop:                        1.44 | expected: 1.44 = N*BLK + N*NBLK + N*GEN
1x async for-loop:                  1.44 | expected: 1.44 = N*BLK + N*NBLK + N*GEN
2x for-loop (separate tasks):       2.08 | expected: 2.08 = 2*N*BLK + 2*N*GEN + NBLK + (N-1)*max(NBLK-(BLK+GEN), 0)
2x async for-loop (separate tasks): 1.59 | expected: 1.59 = 2*N*BLK + N*GEN + NBLK + (N-1)*max(NBLK-BLK, 0) + (N-1)*GEN*(BLK > NBLK)*(NBLK > GEN) + GEN*(NBLK > BLK)*(BLK > GEN)

问题解答 #

1. 慢I/O生成器不是唯一提速场景 #

async for的核心价值在于让生成器的迭代过程能主动让出CPU控制权，只要迭代过程中存在可异步调度的操作（不止慢I/O），都能通过async for实现多任务并行提速。除了生成器内部的慢I/O，还有以下典型场景：

2. 其他提速场景示例：批量异步资源处理 #

假设我们需要从多个远程接口获取数据，再对每条数据做同步CPU计算。用普通for循环会串行等待每个接口响应，而async for配合异步生成器可以让多个请求并行发起，同时利用等待响应的时间处理已返回的数据。

示例代码

import asyncio
import time

# 模拟远程接口请求（异步）
async def fetch_data(item_id):
    await asyncio.sleep(0.5)  # 模拟网络延迟
    return {"id": item_id, "data": item_id * 10}

# 异步生成器：批量发起请求，逐个返回结果
async def async_data_generator(item_ids):
    tasks = [fetch_data(id) for id in item_ids]
    for task in asyncio.as_completed(tasks):
        result = await task
        yield result

# 同步CPU计算
def process_data(data):
    time.sleep(0.2)  # 模拟CPU密集型处理
    return data["data"] * 2

# 使用async for的异步流程
async def async_workflow():
    start = time.perf_counter()
    async for data in async_data_generator(range(5)):
        processed = process_data(data)
        print(f"Processed: {processed}")
    print(f"async for total time: {time.perf_counter() - start:.2f}s")

# 使用普通for的串行流程
async def standard_workflow():
    start = time.perf_counter()
    for id in range(5):
        data = await fetch_data(id)
        processed = process_data(data)
        print(f"Processed: {processed}")
    print(f"standard for total time: {time.perf_counter() - start:.2f}s")

async def main():
    await async_workflow()
    print("---")
    await standard_workflow()

asyncio.run(main())

输出结果

Processed: 0
Processed: 20
Processed: 40
Processed: 60
Processed: 80
async for total time: 1.20s
---
Processed: 0
Processed: 20
Processed: 40
Processed: 60
Processed: 80
standard for total time: 3.50s

提速原理

• 普通for流程：串行执行fetch_data+process_data，总耗时=5*(0.5+0.2)=3.5s。
• async for流程：异步生成器通过asyncio.as_completed同时发起所有请求，当第一个请求返回时就开始处理数据，后续请求在处理当前数据的0.2秒内继续等待，最终总耗时≈最长的请求时间 + 总处理时间=0.5 + 5*0.2=1.5s（实际因调度略有误差）。

这里的关键是异步生成器将批量异步任务的等待和结果处理解耦，async for在迭代时能在等待下一个结果的同时，让事件循环调度其他任务，而普通for必须等待当前迭代的所有步骤完成才能进入下一次循环。

3. 另一场景：流式异步数据消费 #

比如处理来自消息队列的流式数据，异步生成器可以持续监听队列（异步等待新消息），async for能在消息到达时立即处理，同时不阻塞其他异步任务的执行。而普通生成器如果用轮询方式监听队列，会一直占用CPU，且无法和其他任务并行。

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