异步输入输出陷阱：三小时漏洞

2026西湖龙井茶官网DTC发售：茶农直供，政府溯源防伪到农户家 

上周，我的老板要求我加速一个旧的网络爬虫项目。我心想：“没问题——我只需引入 asyncio 进行并发获取，理论上就能将 200 个请求的耗时从 40 秒大幅缩减至 2 秒左右。”然而，在写完第一个版本后，虽然速度确实提升了，但一半的数据丢失了，控制台还充斥着 RuntimeWarning: coroutine was never awaited（运行时警告：协程从未被等待）。更糟糕的是，程序会随机卡死，CPU 占用率为 0%，最终导致超时。在花费三个小时仔细研读文档和源代码后，我终于填平了所有的坑。本文记录了这些易错点以及我最终总结出的最佳实践，助你避免重蹈覆辙。

场景：200 次 API 数据获取，从 40 秒优化至 2 秒

原始的同步代码如下所示——简单但慢得令人痛苦：

import time
import requests

def fetch_all(urls):
    results = []
    for url in urls:
        resp = requests.get(url, timeout=5)
        results.append(resp.json())
    return results

urls = [f"https://api.example.com/item/{i}" for i in range(200)]
start = time.time()
data = fetch_all(urls)
print(f"耗时: {time.time() - start:.2f}s")
# 输出: 耗时: 41.23s

200 个顺序执行的请求耗时超过 40 秒——体验极差。我满怀信心地开始使用 asyncio 重写它。

核心概念：事件循环 + 协程 = 非阻塞并发

asyncio 的工作机制与多线程完全不同。它运行在一个单线程的事件循环中，所有协程都在同一个线程内调度。当协程遇到 IO 等待（网络、磁盘）时