同步 requests 循环抓取时，真正耗时的部分通常不是计算，而是等待网络响应。原文的批量图片下载脚本从几百个链接跑近十分钟，改成 asyncio 并发后压到约二十秒。这个差距来自单线程事件循环：遇到 I/O 等待时协程主动让出控制权，让其他协程先执行。需要先说明适用边界：asyncio 适合爬虫、API 调用、数据库读写等 I/O 密集型任务；CPU 密集型计算应交给 multiprocessing 或 C 扩展，协程本身不会让计算变快。
一、环境与最小可运行验证
asyncio 在 Python 3.4 就已存在，但真正省去手动创建和关闭事件循环的 asyncio.run() 从 Python 3.7 才引入。建议至少使用 Python 3.8，原文更推荐 3.10 或 3.11，性能和错误提示更友好。Windows 安装时如果没勾选 Add Python to PATH，终端会提示 Python was not found；可重新运行安装包选择 Modify，在 Optional Features 中确认 Add Python to environment variables。Linux 和 macOS 先用 python3 --version 检查版本，必要时用系统包管理器升级，例如 Ubuntu 执行：
- sudo apt update
- sudo apt install python3 python3-pip
VSCode 中先安装官方 Python 扩展，再用 Ctrl+Shift+P 执行 Python: Select Interpreter，选中刚装好的解释器。项目内建议建立虚拟环境：
- python -m venv venv
- source venv/bin/activate
Windows 下激活命令为 venv\Scripts\activate。解释器选错是异步代码报错的常见来源，务必确认 VSCode 右下角路径与终端中的 Python 一致。环境就绪后，用下面的最小脚本验证：
- import asyncio
- async def main():
- print('hello')
- await asyncio.sleep(1)
- print('world')
- asyncio.run(main())
运行后应先输出 hello，约 1 秒后输出 world。这里 asyncio.sleep(1) 与 time.sleep(1) 的关键区别是：前者会释放事件循环，让其他协程获得运行机会。
二、协程、事件循环与 Task 的关系
async def 定义的是协程函数。调用它不会立即执行函数体，而是返回协程对象；只有被事件循环调度后，函数体才真正运行。await 用来等待可等待对象，包括协程对象、Task、Future 以及 asyncio.sleep() 的返回值。执行到 await 时当前协程暂停，事件循环转去处理其他任务，等对象就绪后再恢复。
asyncio.run() 会创建事件循环、执行传入的协程并关闭循环。它每次调用都会创建新循环，因此不要在同一个程序里反复创建循环并期望状态保留。调试多协程代码时，最好只在入口调用一次 asyncio.run()，把所有协程组织进一个 main 协程。
要让多个协程真正并发，需要用 Task 包装。asyncio.create_task() 会把协程注册到事件循环，等待调度：
- import asyncio
- async def worker(name, delay):
- print(f'{name} start')
- await asyncio.sleep(delay)
- print(f'{name} end')
- async def main():
- task1 = asyncio.create_task(worker('task1', 2))
- task2 = asyncio.create_task(worker('task2', 3))
- await task1
- await task2
- asyncio.run(main())
由于 task1 睡眠期间事件循环会调度 task2，所以 task2 start 往往会在 task1 结束前打印。create_task 必须在事件循环运行中调用，也就是只能在协程里创建，不能在最外层直接调用。多个独立任务还可以用 asyncio.gather() 聚合等待，并拿到返回值列表。
三、同步下载与异步下载实测对比
原文用 httpbin.org/delay/1 模拟每个请求等待 1 秒，发起 10 个请求。同步版本如下：
- import requests
- import time
- urls = ['https://httpbin.org/delay/1'] * 10
- def fetch_sync(url):
- resp = requests.get(url)
- return resp.status_code
- start = time.time()
- for url in urls:
- fetch_sync(url)
- print(f'同步耗时: {time.time() - start:.2f}s')
10 个请求串行完成，总耗时约 10 秒。换成 asyncio 加 aiohttp，因为 requests 是同步库，不能直接塞进 asyncio：
- import asyncio
- import aiohttp
- import time
- urls = ['https://httpbin.org/delay/1'] * 10
- async def fetch_async(session, url):
- async with session.get(url) as resp:
- return resp.status
- async def main():
- async with aiohttp.ClientSession() as session:
- tasks = [fetch_async(session, url) for url in urls]
- results = await asyncio.gather(*tasks)
- print(results)
- start = time.time()
- asyncio.run(main())
- print(f'异步耗时: {time.time() - start:.2f}s')
原文实测同步版为 10.03 秒，异步版为 2.17 秒。异步版不是严格的 1 秒，是因为 DNS 解析、连接建立、TLS 握手等仍有额外开销。这里的 async with 用于管理 aiohttp.ClientSession()，退出时会释放连接池资源；如果只打开不关闭，长时间运行可能出现 Too many open files。任务数量特别大时不要一次性创建几千个请求，可以每批 200 个分批处理，避免内存和文件描述符被打满。
四、异步上下文管理器、异步迭代器与超时取消
async with 依赖 __aenter__ 和 __aexit__ 两个方法，适合封装需要异步清理的资源。自定义类的结构如下：
- import asyncio
- class AsyncConnection:
- async def __aenter__(self):
- print('opening connection')
- await asyncio.sleep(0.1)
- return self
- async def __aexit__(self, exc_type, exc, tb):
- print('closing connection')
- await asyncio.sleep(0.1)
- async def main():
- async with AsyncConnection() as conn:
- print('using connection')
即使 with 块内抛出异常，__aexit__ 也会被调用，比手写零散的 close 语句更稳妥。异步迭代器则用于逐条获取异步数据，例如分页抓取：
- async def fetch_pages():
- for page in range(1, 6):
- await asyncio.sleep(0.5)
- yield f'page {page} data'
- async def main():
- async for data in fetch_pages():
- print(data)
async for 要求对象实现 __aiter__ 和 __anext__。实际开发中更多是使用异步 Redis 的 scan_iter 等库自带迭代器，此时 async for 是正确写法。
超时控制用 asyncio.wait_for() 给协程或任务设置最大等待时间：
- try:
- result = await asyncio.wait_for(slow_operation(), timeout=3)
- except asyncio.TimeoutError:
- print('超时了')
超时后 wait_for 会尝试取消任务，但取消只在协程执行到下一个 await 点时抛出 CancelledError。如果协程内部调用同步阻塞库，就无法优雅取消。需要清理资源时，可在协程中捕获 asyncio.CancelledError，完成清理后再 raise：
- async def controlled_task():
- try:
- await asyncio.sleep(10)
- except asyncio.CancelledError:
- print('cleanup')
- raise
另外，asyncio.gather() 默认不会因为某个任务抛异常就取消其他任务。如果需要异常不影响其他任务，可传入 return_exceptions=True，让异常作为结果条目返回，再统一判断。
五、五个常见错误及排查思路
1. RuntimeError: no running event loop。常见于普通同步函数里直接调用 asyncio.create_task()。create_task 需要当前线程存在正在运行的事件循环，解决方式是把创建动作放进协程，或用 asyncio.run_coroutine_threadsafe() 把协程投递给另一个线程的事件循环。排查时先确认这行代码是在哪条协程、哪个线程中执行。
2. 协程忘记 await。coro() 只创建协程对象，函数体不会执行，必须 await coro()。asyncio.gather(coro(), coro()) 会内部包装成 Task，所以不会报错，但把协程对象传给其他函数时可能一直不执行。排查时可在每个协程第一行打印 enter 日志，确认哪些协程真正进入。
3. 在协程里写 time.sleep。time.sleep 是同步阻塞，休眠期间不会让出事件循环，其他协程全部无法运行，多个任务总耗时仍会变成 1 秒乘任务数。检查所有可能阻塞的调用，包括 time.sleep、requests.get、普通 open().read()。没有异步版本的库可以用 asyncio.to_thread() 丢到线程池，避免阻塞循环。
4. gather 返回值顺序和异常处理。gather 返回列表顺序与传入任务顺序一致，不是完成时间顺序。需要按完成顺序处理结果时应使用 asyncio.as_completed()。默认情况下任一任务抛异常，gather 会立即向上传播，其他任务仍会继续执行，但结果已经拿不到；需要保留全部结果时传 return_exceptions=True。
5. asyncio.run 与 loop.run_until_complete 混用。老教程常见 loop = asyncio.get_event_loop() 后 loop.run_until_complete(main())。新代码推荐 asyncio.run(main())，它会自动创建和关闭循环。两种方式混用会遇到事件循环已关闭的报错。排查思路是只在程序最顶层调用一次 asyncio.run()，轮询任务和监听信号都放到 main 协程内部完成。
六、爬虫与数据采集中的落地结构
批量抓取时应限制并发数，避免把目标站点打崩或触发封禁。可以用 asyncio.Semaphore(20) 控制同时打开的连接数，把单次请求包在 async with sem 中。原文给出的启动框架如下：
- async def main():
- sem = asyncio.Semaphore(20)
- async with aiohttp.ClientSession() as session:
- tasks = [fetch_with_retry(session, sem, url) for url in urls]
- results = await asyncio.gather(*tasks, return_exceptions=True)
这个结构便于后续加入限速、重试和超时。爬虫场景可以按批构造任务，用异步队列去重和限速；量化行情采集场景可以异步并发请求数百只股票的分钟级数据，数据回来后再统一进入策略计算。如果策略计算是纯 CPU 密集，就把重计算交给多进程，形成“异步拿数 + 多进程算数”的组合。
调试异步代码不要只靠 print。可以用调试器给每个 Task 单独打断点，或用 asyncio.all_tasks() 打印当前所有未完成任务，观察是否有协程被卡住。理解协程的挂起与恢复机制后，再看 asyncio 生态中的异步数据库、异步 HTTP 客户端和异步队列，会更容易判断哪些调用会阻塞事件循环。 |
发表于 
发表于