先分享一个真实案例。 上个月，我接到一项任务：编写一个爬虫程序，需要抓取一万个网页。每个网页内部又包含几十张图片链接，最终还需将这些图片全部下载到本地。 听起来很简单，对吧？直接用 `requests` 库循环一万次，再在内部循环几十次，就搞定了。 但仔细推敲，这种方案并不可行。一万个网页，每个网页几十张图片，累计起来是几十万次网络请求。如果采用同步方式一个一个等待，估计程序跑完时，下一次版本迭代都已经上线了。 于是自然想到了异步方案。`asyncio` 和 `aiohttp` 全部安排上。 代码写完后，一运行——速度极慢。几乎和同步版本没有区别，完全没有发挥异步编程应有的并发优势。 让人困惑的是，问题究竟出在哪里？ 折腾了整整一个晚上，翻阅了无数篇技术帖子，最终发现原因藏在一个完全被忽略的细节中。 今天我就把这个坑完整地剖析给你听。保证听完之后，你不仅知道如何避开它，还能真正理解异步循环嵌套的本质。 ## 先搭建一个场景 我们用一个简化的小例子来说明。假设你需要从三个网站上抓取数据，每个网站都需要先请求 `page` 接口（耗时1秒），然后根据返回的结果再请求 `detail` 接口（也耗时1秒）。 同步写法非常直观： import time

def fetch_page(site):
time.sleep(1) # 模拟网络请求
return f"{site} 的数据"

def fetch_detail(site):
time.sleep(1)
return f"{site} 的详细信息"

def main():
sites = ["site_a", "site_b", "site_c"]
for site in sites:
page = fetch_page(site)
detail = fetch_detail(site)
print(page, detail)

start = time.time()
main()
print(f"耗时: {time.time() - start:.2f}秒")
运行一下，总耗时大约6秒。每个站点需要2秒，三个站点累计6秒。符合预期。 那么异步版本呢？在理想情况下，三个站点的请求可以同时进行，总耗时只需要2秒左右。 我们来实现异步版本： import asyncio

async def fetch_page(site):
await asyncio.sleep(1)
return f"{site} 的数据"

async def fetch_detail(site):
await asyncio.sleep(1)
return f"{site} 的详细信息"

async def process_site(site):
page = await fetch_page(site)
detail = await fetch_detail(site)
return page, detail

async def main():
sites = ["site_a", "site_b", "site_c"]
tasks = [process_site(site) for site in sites]
results = await asyncio.gather(*tasks)
for result in results:
print(result)

start = time.time()
asyncio.run(main())
print(f"耗时: {time.time() - start:.2f}秒")
这个版本运行耗时约为2秒，完美实现了并发。 这个例子看起来非常简单，对吧？但恰恰是在这个基础上，一旦嵌套了一层循环，问题就悄然出现了。 ## 我真实代码的结构 当时我的代码大致是这样的： async def fetch_page(site, page_num):
await asyncio.sleep(0.1) # 模拟请求
return f"{site} 第{page_num}页的数据"

async def fetch_images(page_data):
await asyncio.sleep(0.05) # 模拟请求图片
return [f"image_{i}" for i in range(3)]

async def process_site(site):
all_images = []
# 外层循环：该站点的每一页
for page_num in range(1, 11): # 假设每个站点有10页
page_data = await fetch_page(site, page_num)
# 内层循环：这一页的每一张图片
images = await fetch_images(page_data)
all_images.extend(images)
return all_images

async def main():
sites = ["site_a", "site_b", "site_c"]
tasks = [process_site(site) for site in sites]
results = await asyncio.gather(*tasks)
乍一看似乎没有问题？外层循环处理每个站点，内层循环处理每个站点的每一页，而每页内的图片请求又是异步执行的。这不挺好的吗？ 但实际运行后发现：三个站点之间确实是并发的，可每个站点内部的10页却是顺序执行的——先请求第1页，等返回后再请求第1页的图片，然后才开始第2页，再等图片，再第3页…… 这相当于三个站点各排成一队，一页一页依次处理。完全没有利用到“同一页里的多张图片可以同时下载”这个优化机会。 更糟糕的是，如果每个站点有100页，每页有50张图，那么这种顺序执行的问题会被放大100倍。 这个问题相当隐蔽，不容易一眼看穿。直到我在纸上画出执行顺序图，才恍然大悟。 ## 画出执行顺序你就明白了 我们手动模拟一下执行过程。假设只有两个站点，每个站点只有两页，每页包含两张图片。 当时那段代码的执行顺序如下： 站点A：请求第1页 → 等待 → 拿到数据 → 请求图1 → 等待 → 请求图2 → 等待 → 存图
站点A：请求第2页 → 等待 → 拿到数据 → 请求图1 → 等待 → 请求图2 → 等待 → 存图
站点B：请求第1页 → 等待 → 拿到数据 → 请求图1 → 等待 → 请求图2 → 等待 → 存图
站点B：请求第2页 → 等待 → 拿到数据 → 请求图1 → 等待 → 请求图2 → 等待 → 存图
每一个“等待”的位置，CPU其实都处于空闲状态，但程序就是不肯去做其他事情，非要等这个请求返回。 这就是问题的核心：**`await` 会挂起当前这个异步函数，但它仅仅挂起自己，并不会影响同一层级上的其他任务**。 等等，这句话概念有点绕。我用大白话再解释一遍： 当你在一个异步函数里写下 `await something()`，这个函数就会停在这里，等待 `something()` 完成。但这并不意味着整个程序都停止了——程序可以转而去执行别的异步任务，比如另一个站点的任务。 所以在上面的代码中，`process_site('site_a')` 这个任务在等待第1页返回时，程序确实可以去处理 `process_site('site_b')`。这一点是好的，因此三个站点之间实现了并发。 问题在于：在同一个 `process_site` 任务内部，`for` 循环里的每一次 `await` 都会让这个任务停下来，直到本次请求完成，才会进入下一次循环。内层循环同样如此。 因此，每个站点内部的所有请求都是串行的。 ## 正确的做法是什么？ 如果你希望一个站点内部的多个请求也能并发执行，你需要将这些独立的请求批量收集起来，然后使用 `asyncio.gather` 或 `asyncio.wait` 一次性发出。 以图片下载为例，正确的做法应该是：先获取当前页面的所有图片链接，然后一次性创建所有图片的异步任务，同时等待它们全部完成。 代码大致如下： async def process_site_correct(site):
all_images = []
for page_num in range(1, 11):
page_data = await fetch_page(site, page_num)
# 先提取这一页的所有图片链接
image_urls = extract_image_urls(page_data)
# 关键点：一次性创建所有图片任务，并发执行
image_tasks = [fetch_image(url) for url in image_urls]
images = await asyncio.gather(*image_tasks)
all_images.extend(images)
return all_images
修改后的执行顺序如下： 站点A：请求第1页 → 等待
（等待期间，站点B可以处理自己的任务）
第1页返回 → 同时请求该页的所有图片（假设10张图同时发送请求）
等待所有图片返回 → 然后继续第2页
图片下载部分从此由串行变成了并发。 但这里还有进一步的优化空间：页面请求本身是否也能并发？比如一个站点有10页，是否可以同时请求这10页？ 可以。但需要注意的是，同时请求10页可能会给目标服务器带来较大压力，也可能导致你自己的网络连接数急剧上升。合理控制并发数是一个单独的话题，今天暂不展开。 ## 更隐蔽的坑：嵌套循环里的 gather 我们再深入一层。假设每个站点的每一页返回的数据中，不仅包含图片链接，还需要调用额外的 API（例如每张图片需要请求一个评论接口）。 此时代码可能会变成这样： async def fetch_image_with_comments(image_url):
image_data = await fetch_image(image_url)
comments = await fetch_comments(image_url)
return {"image": image_data, "comments": comments}

async def process_page(page_num):
page_data = await fetch_page(page_num)
image_urls = extract_urls(page_data)
# 这里看起来是并发的
tasks = [fetch_image_with_comments(url) for url in image_urls]
results = await asyncio.gather(*tasks)
return results
这个设计看起来没问题吧？每个 `fetch_image_with_comments` 内部确实是串行的（先等图片，再等评论），但不同图片之间实现了并发。 这已经是一个不错的改进。 但如果你写出这样的代码： # 错误示范
async def fetch_image_with_comments_wrong(image_url):
# 内部又套了一层循环？或者用了 gather 却忘了 await？
tasks = [fetch_image(image_url), fetch_comments(image_url)]
# 这里没有 await，返回的是一个协程对象，而不是结果
return asyncio.gather(*tasks) # 注意：这里没有 await
你会在某个地方发现结果不符合预期，或者更糟糕——程序根本没有执行这些请求，因为你返回的是一个未被调度的协程对象。 这属于另一个经典错误：`asyncio.gather` 返回的是一个 awaitable 对象，你必须使用 `await` 来等待它，或者用 `asyncio.run` 来驱动，否则它不会真正执行。 ## 调试方法：打日志看时间 如果你不确定自己的异步代码是否真的实现了并发，最简单的办法就是添加时间戳日志。 import time

async def fetch_with_log(name, delay):
start = time.time()
print(f"[{start:.3f}] 开始 {name}")
await asyncio.sleep(delay)
end = time.time()
print(f"[{end:.3f}] 结束 {name}，耗时 {end-start:.2f}秒")
return name

async def test_serial():
print("串行版本：")
for i in range(3):
await fetch_with_log(f"任务{i}", 0.5)

async def test_concurrent():
print("并发版本：")
tasks = [fetch_with_log(f"任务{i}", 0.5) for i in range(3)]
await asyncio.gather(*tasks)

# 运行后你会看到明显区别
# 串行：开始时间依次相差0.5秒
# 并发：三个任务的开始时间几乎相同
这个小技巧我已经用了无数次。每当你怀疑某个地方的循环是否串行时，在关键操作前后加上日志，看看开始时间是否挤在一起。 如果开始时间是一条直线依次出现，那就是串行；如果几乎同时打印出来，那就是并发。 ## 几条简单规则 经过大量实践，我总结出以下几条规则，供你参考： **规则1：看到 `await` 在循环里，就要保持警惕** `for` 循环内部如果直接 `await` 一个异步函数，那么这个循环一定是串行的。除非你刻意需要串行，否则应该考虑先收集任务，再用 `gather` 集中执行。 **规则2：明确“谁和谁可以并发”** - 不同站点之间：可以并发 - 同一个站点的不同页面：如果服务器能承受，可以并发 - 同一个页面里的不同图片：可以并发 - 同一张图片的下载和评论请求：通常不能并发（存在依赖关系） **规则3：`gather` 不是万能药，它只是“同时等待”** 很多人以为用了 `gather` 就自动并发了。其实 `gather` 做的事情很简单：将你传入的多个协程任务同时调度起来，然后等待它们全部完成。但前提是这些任务本身是独立的。 如果你传给 `gather` 的是一系列 `[fetch_page(1), fetch_page(2), fetch_page(3)]`，这三个请求会同时发出，效果很好。 但如果你传给 `gather` 的是一系列 `[process_page(1), process_page(2), process_page(3)]`，而每个 `process_page` 内部又是串行的，那么 `gather` 也无法拯救性能。 **规则4：异步不等于自动并行** 这是最容易误解的一点。`async/await` 提供的仅仅是“在等待时不被阻塞”，而非“自动将循环拆成多线程”。并发需要你显式地使用 `gather`、`create_task`、`wait` 等工具来组织。 ## 回到那个爬虫 最后，我那个爬虫改成了以下形式： async def process_site_optimized(site):
# 先获取该站点所有需要抓取的页面列表
page_tasks = [fetch_page(site, page_num) for page_num in range(1, 101)]
# 同时请求所有页面（用 semaphore 控制并发数）
pages_data = await limited_gather(page_tasks, max_concurrent=10)

# 收集所有图片 URL
all_image_tasks = []
for page_data in pages_data:
image_urls = extract_image_urls(page_data)
all_image_tasks.extend([fetch_image(url) for url in image_urls])

# 同时下载所有图片（同样限制并发）
images = await limited_gather(all_image_tasks, max_concurrent=20)
return images

来源：https://developer.aliyun.com/article/1739560

Python

上一篇无需地图也能规划公交路线：首个端到端大规模数据集 下一篇服务器RAID5光纤存储故障数据恢复案例解析

本站内容用于信息整理与展示，如有侵权或内容问题请及时联系处理。