/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}
说实话，我第一次自己写爬虫的时候，真的被“慢”整破防了。那时候我用 Python 写了一个批量下载图片的小脚本，requests 一个个去请求，跑完几百个链接花了快十分钟。我一开始以为是电脑太差，后来才发现我陷入了同步阻塞的坑里，完全没用到 Python 异步编程的能力。后来我花了一个周末把 asyncio 库啃下来，同样的任务从十分钟直接压到二十秒，这个体验真的相当震撼。
这篇内容不是照抄官方文档，而是从我自己的学习路径出发，把 Python 异步编程入门最核心的 asyncio 概念、使用逻辑、常见坑和真实场景串一遍。无论你是刚学 Python 的新手，还是已经在写爬虫、量化交易数据采集脚本但觉得速度太慢的人，这篇文章应该都能帮你在短时间内理清思路，把 asyncio 用起来。
1. 为什么需要异步：网络请求慢的本质是“等待”
1.1 同步代码到底在等什么
在讲 asyncio 之前，必须先搞清楚一个问题：你的代码时间到底花在哪了。以最经典的 requests 请求为例：
import requests
import time
def download_one(url):
resp = requests.get(url)
return len(resp.content)
start = time.time()
for url in urls:
download_one(url)
print(f"total: {time.time() – start:.2f}s")
这段代码在循环里一个接一个地请求。表面上看是“执行请求”占用了时间，但如果你真的去分析，会发现 99% 的时间都花在了等待网络响应上，也就是
resp = requests.get(url)
这一行卡住不动。CPU 在这个阶段几乎什么都没干，既没算数据，也没写文件，就是傻等着网卡把数据传回来。
这种等待叫作 I/O 等待。不只是网络请求，数据库查询、文件读写、子进程调用，都会有这种等待。同步代码的问题在于，一个人等的时候，其他活也干不了。就像银行只有一个窗口，一个人办业务办十分钟，后面所有人都只能干站着。
1.2 多线程能解决，但为什么不是首选
很多人的第一反应是：那用多线程不就行了吗？确实可以，Python 的
threading
模块能让多个请求并行等待，但代价也很明显。每个线程都有不小的创建和切换成本，线程一多，内存和上下文切换开销都会上来。
更重要的是，Python 有一个全局解释器锁 GIL。GIL 导致同一时刻只有一个线程能执行 Python 字节码，多个线程并不是真的在并行计算。虽然 I/O 等待时 GIL 会被释放，但线程调度的不确定性、共享资源的锁管理，让多线程写起来很累，调试起来更累。
1.3 协程的切换逻辑：更像“自己安排顺序”
asyncio 走的是另一条路：单线程内，用一个事件循环来调度多个协程。协程是一种可以暂停和恢复的函数。遇到 I/O 等待时，协程主动把控制权交还给事件循环，说“我先等着，别人有活你先去忙”，等数据回来了，再从暂停的地方继续执行。
这种切换是协作式的，没有抢占，也没有线程切换的系统开销。我把协程之间的切换类比成餐厅服务员手里同时记着几桌客人的菜单：你不需要每一个菜都站在锅边等，下单、传菜、结账之间穿插着来，效率自然高。Python 异步编程的核心就是用 asyncio 库帮助我们在单线程里实现这种高并发。
注意：asyncio 适合的是 I/O 密集型任务，比如爬虫、API 调用、读写数据库。如果是 CPU 密集型任务，比如大量计算、视频编码、复杂的数值模拟，协程并不会让它更快，这时候应该用
multiprocessing
或者把它丢给 C 扩展去处理。
2. 先把环境弄对：Python 版本与 VSCode 配置
2.1 选择 Python 3.7 以上版本
asyncio 在 Python 3.4 就有了，但早年 API 还比较粗糙。真正好用的
asyncio.run()
是从 Python 3.7 引入的，它替我们省掉了手动创建和关闭事件循环的麻烦。所以我建议你至少用 Python 3.8 以上，最好直接装 3.10 或者 3.11，性能和错误提示都会友好很多。
Windows 用户安装时有一个很容易踩的坑：安装包下载好了，下一步下一步点完，打开终端输入
python
，结果提示：
Python was not found; run without arguments to install from the Microsoft Store…
出现这个是因为安装时没有勾选“Add Python to PATH”。正确的做法是安装第一页时把最底部的这个选项勾上，再点 Install Now。如果你已经装好了但没勾选，也可以重新运行安装包，选择 Modify，然后在 Optional Features 页面确认把 Add Python to environment variables 选上。
Linux 和 macOS 用户一般系统里自带 Python 3，但版本可能比较老。建议用
python3 –version
看一下，如果版本太低，可以用系统包管理器安装新的版本，比如 Ubuntu 上执行：
sudo apt update
sudo apt install python3 python3-pip
macOS 上如果装了 Homebrew：
brew install python
装完最重要的一步，是确认你能在终端里进入 Python 交互环境，并且看到版本号。这一步过了，后面代码才不会被环境问题卡住。
2.2 VSCode 里的 Python 环境配置
VSCode 是入门 Python 最友好的编辑器之一，但环境配置有几个细节要注意。首先要装官方 Python 扩展，这个没得商量。装完之后，打开一个文件夹，按
Ctrl+Shift+P
，输入“Python: Select Interpreter”，选择你刚刚装好的那个 Python 版本，而不是系统里其他诡异路径下的解释器。
如果你打算用虚拟环境，建议现在就养成习惯。在项目根目录下打开终端：
python -m venv venv
然后运行
source venv/bin/activate
（macOS/Linux）或者
venv\\Scripts\\activate
（Windows）。VSCode 检测到虚拟环境后，会在右下角显示当前解释器路径。我看到很多人写异步代码报错，最后发现是系统里有多个 Python，VSCode 选了解释器 A，终端里用的却是解释器 B，这种低级错误非常浪费时间。
2.3 快速验证 asyncio 能否运行
环境配置好后，在 VSCode 里新建一个
test_async.py
，写入最基础的代码：
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)
最大的区别是它会释放事件循环，给其他协程运行的机会。这个区别很重要，后面讲坑的时候会专门展开。
3. asyncio 三大基础概念：协程、事件循环、任务
3.1 async def 和 await：协程的两块基石
在 asyncio 的世界里，一切都是围绕协程展开的。定义协程的方式很简单，就是在普通函数前面加一个
async
：
async def my_coroutine():
return 1
调用这个函数时，并不会立刻执行函数体，而是返回一个协程对象。协程对象只有被事件循环调度到，函数体里的代码才会真正运行。这和使用普通函数的直觉差别很大，很多人写着写着就丢了个
await
，代码一点反应都没有。
await
的作用是等待一个可等待对象。什么叫可等待？协程对象、
Task
对象、
Future
对象，以及
asyncio.sleep()
这种返回的对象，都可以被
await
。当代码执行到
await
时，当前协程会暂停，事件循环会去看别的协程有没有事情要做，等这个可等待对象就绪了，再回来继续执行。
我之前对
await
一直没办法理解透彻，后来想到了一个比喻：
await
相当于你给朋友发消息说“你到了告诉我”，发完你没傻傻地盯着手机，而是先去洗碗、拖地，等朋友回复了你再继续聊天。这个“先去干别的”的机制，就是异步的本质。
3.2 事件循环：背后的大总管
事件循环可以理解成一个无限循环，它持续地检查任务列表里哪些协程已经准备好执行。准备好就执行，碰到
await
就挂起来，等 I/O 事件回来后再标记为可执行。
Python 3.7 之后，我们最常见的入口是
asyncio.run()
。它会完成三件事：创建事件循环、执行传入的协程、关闭事件循环。比如刚才的
test_async.py
里，
asyncio.run(main())
就是在里面跑了一个事件循环，调度 main 这个协程。
需要注意，
asyncio.run()
每次调用都会创建一个新的事件循环，所以不能在同一进程里反复创建不俗循环然后期望状态保留。调试多协程代码时，最好在入口只调用一次
asyncio.run()
，把所有协程都组织到一个 main 协程里。
3.3 Task：让并发真正发生
我们刚说了
await
一个协程，比如
await coro()
，这个协程会按顺序执行完，期间虽然会让出控制权，但本质上你就是在等它。如果想要多个协程同时跑，就得用
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())
这里的
create_task
不会立刻执行协程，而是把它注册到事件循环里，等待调度。当你
await task1
时，事件循环发现 task2 也已经注册了，就会穿插执行。所以上面代码里
task2 start
往往在
task1 sleep
期间就打印出来了。
还有一个常用的组合是
asyncio.gather()
，它可以把一堆任务聚合成一个整体来等待：
await asyncio.gather(task1, task2)
用
gather
的好处是不用手动逐个 await，还能拿到所有协程的返回值列表。如果你的场景是多个请求互相独立，
gather
就是最顺手的工具。
提示：创建 Task 的方式不唯一。老代码里经常看到
loop.create_task()
或
asyncio.ensure_future()
，现在主流写法就是
asyncio.create_task()
。它必须在事件循环运行中调用，也就是只能在协程里创建，不能在最外层直接
create_task
。
4. 第一份并发代码：同步下载 vs 异步下载实测对比
4.1 同步版代码
光讲概念没用，来看一个能真实感知到差距的例子。我模拟一个批量请求场景，用
httpbin.org/delay/1
这个接口，它会等待 1 秒后再返回，方便我们观察耗时。
先写同步版本：
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")
运行结果很朴素，每个请求等 1 秒，10 个请求串行完成，总耗时约 10 秒。
4.2 异步版代码
然后换成 asyncio 加 aiohttp。
aiohttp
是异步 HTTP 客户端，因为原来的
requests
是同步库，没办法直接塞进 asyncio 里，所以我们用
aiohttp
重写请求部分：
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")
如果网络条件正常，异步版本的耗时大约在 1 秒到 3 秒之间，而不是 10 秒。区别就是这 10 个请求在同一个事件循环里并发执行，整体等待时间等于最慢的那个请求，而不是所有请求的总和。
4.3 实测结果解读和参数选择
我第一次跑这个对比时，同步版 10.03 秒，异步版 2.17 秒。为什么不是严格的 1 秒？因为 aiohttp 的连接建立、TLS 握手、DNS 解析这些环节仍然会有额外开销，竞态也不可避免。但相比同步版本，速度提升是非常明显的。
这里的
async with
也值得说一下。
aiohttp.ClientSession()
需要在使用后关闭，
async with
是一个异步上下文管理器，它能保证在退出时干净地释放连接池资源。异步代码里的资源管理往往比同步代码更需要注意，因为你如果随手打开 session 却不关闭，连接池里的连接会一直占着，程序跑久了就会出现“Too many open files”之类的系统错误。
还有一个细节：
asyncio.gather(*tasks)
传入的是一个可迭代的任务列表。如果任务数量特别多，比如一次跑几千个请求，不建议一次性全部创建，否则内存和文件描述符会被瞬间打满。可以分批处理，每批 200 个请求，跑完再开下一批。
5. 进阶用法：异步上下文管理器、异步迭代器、超时与取消
5.1 async with 让资源释放不再看心情
前面提到的
aiohttp.ClientSession()
只是一个例子。在 asyncio 生态里，很多资源都实现了异步上下文管理器，比如异步数据库驱动的连接、Redis 连接、文件锁等。
async with
既能保证退出时执行异步清理逻辑，又比手动
await conn.close()
更安全。
如果你自己写一个类想支持
async with
，需要实现
__aenter__
和
__aexit__
两个魔法方法。比如一个简单的异步数库连接：
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")
用处是让代码更清晰，避免写过散的 close 语句，尤其当中间逻辑抛出异常时，
__aexit__
也一样会被调用，比 try/finally 写起来舒服。
5.2 async for 处理流式数据
普通的
for
循环迭代一个同步可迭代对象，如果迭代过程中涉及到网络或 I/O 等待，普通循环会阻塞事件循环。异步迭代器则可以每次获取下一个元素时 yield 出控制权，等数据到了再继续。
这个在爬虫里很常见，比如从一个异步队列里取 URL，或者对 API 分页返回的数据逐页处理：
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
就是唯一正确写法。
5.3 超时控制与任务取消
异步程序里最常见的问题之一：某个网络请求一直不返回，整个 gather 卡在那，谁也不结束。必须设置超时。
asyncio.wait_for()
可以给一个协程或任务设置最大等待时间：
try:
result = await asyncio.wait_for(slow_operation(), timeout=3)
except asyncio.TimeoutError:
print("超时了")
超时后，
wait_for
会尝试取消被等待的任务。但取消这件事有点微妙，不是随便就能停的。只有在协程执行到下一个
await
点时，
CancelledError
才会被抛出。如果协程里面用的是同步阻塞库，那就无法优雅取消了。
任务取消的自定义可以在协程里捕获
asyncio.CancelledError
，做清理工作后再继续抛出：
async def controlled_task():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
print("cleanup")
raise
一个比较容易忽视的点是，
asyncio.gather()
默认不会因为某个任务抛异常就取消其他任务。如果你想实现“一个失败全部取消”，需要传递
return_exceptions=True
或手动组织逻辑。这也是我在实际写批量任务时踩过的一个坑。
6. 新手最常见的五个坑及排查思路
6.1 RuntimeError: no running event loop
这个错误常见于在非协程的普通函数里调用
asyncio.create_task()
。比如你写了一个同步函数，里面想丢一个任务进去，直接
asyncio.create_task(…)
，大概率会报
no running event loop
。
原因是
create_task
需要当前线程有正在运行的事件循环。解决方法是，把这个动作放在协程里，或者用
asyncio.run_coroutine_threadsafe()
把协程投递给另一个线程的事件循环。我见过很多人写定时任务时栽在这里，排查思路就是先确认“这行代码是什么时候、在哪条协程里执行”。
6.2 协程忘记 await，一切静悄悄
假如你写着写着，发现某个协程函数没有被打印，大概率是少了
await
。比如：
task1 = coro() # 只是创建了协程对象，函数体没执行
await task1 # 这样才对
更隐蔽的情况是使用了
asyncio.gather(coro(), coro())
，这里传的是协程对象，gather 会内部包装成 task，所以不会报错，但如果你把协程对象当作参数传给别的函数，可能就一直不被执行。排查时我习惯在每个协程的第一行加一个
print("enter xxx")
，然后看日志里到底哪些协程真正进入了执行。
6.3 在协程里写 time.sleep，循环卡死
time.sleep()
是同步阻塞的。它在休眠期间不会让出事件循环，也就是说其他协程全都跑不了。我在学习阶段写过一个类似代码：
async def bad_worker():
time.sleep(1)
十来个任务一跑，总耗时变成了 1 秒乘任务数，和同步没区别。排查思路：检查所有可能导致阻塞的调用，包括
time.sleep
、
requests.get
、普通
open().read()
。如果这些库没有异步版本，可以用
asyncio.to_thread()
把它们丢到线程池里运行，避免阻塞循环。
6.4 gather 返回值顺序和异常处理
asyncio.gather()
返回的结果列表顺序，与传入任务的先后顺序一致，而不是完成时间顺序。这点很多人误解，以为先完成的任务结果在前。如果你需要按完成顺序处理结果，不要用 gather，要用
asyncio.as_completed()
。
异常也要仔细看。默认情况下，gather 里任何一个任务抛异常，整个 gather 会立刻向上传播，其他任务仍然会继续执行，但已经拿不到它们的结果了。如果你希望异常不影响其他任务，可以把
return_exceptions=True
传进去，这样异常会作为条目出现在结果里，再自己统一判断。
6.5 asyncio.run 和 loop.run_until_complete 混用
网上资料新旧混杂，老教程里常见写法是：
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
新代码里
asyncio.run(main())
更推荐，因为它会自动创建和关闭事件循环。如果两种方式混合在同一个程序里，比如先
asyncio.run()
后来又想拿同一个 loop 去
run_until_complete
，会遇到事件循环已关闭的报错。
排查思路：尽量在程序最顶层只调用一次
asyncio.run()
，所有协程都通过这一条线进入。如果需要轮询任务、监听信号，也应该在 main 协程内部完成，不要在外面反复创建 loop。
7. 实际场景应用：从爬虫到量化数据采集
理论说再多都不如落到具体项目里。我主要用 asyncio 做两类事情：批量爬虫和金融行情数据采集。
爬虫场景的思路很简单：把待抓取的 URL 列表分批构造协程任务，用 aiohttp 请求，用异步队列去重和限速。比如要对几千个商品页做价格监控，同步请求可能要半小时，异步批次请求可以缩到几分钟。重要的是在每个请求之间留一个很小的信号量控制并发数，避免把目标网站打崩，也避免自己 IP 被封。实现的时候可以用
asyncio.Semaphore(20)
，把每次请求包裹在
async with sem:
里面，限制同时打开的连接数。
量化交易策略代码里，同步拉取数百只股票的分钟级行情，同样会遇到 I/O 等待。用 asyncio 并发请求行情接口，数据回来后再统一进入策略计算模块，比单线程逐个请求快很多。策略计算如果是纯 CPU 密集，再把重计算部分交出去，这就是一个典型的“异步拿数 + 多进程算数”组合模式。
这类项目的启动框架可以复用：
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)
这样结构清晰，后续加限速、重试、超时都方便。
根据我个人经验，学习 asyncio 最大的障碍不是语法，而是思维方式。同步代码习惯了一步接一步执行，异步代码则需要你接受“任务会被挂起，再被恢复”的事实。刚开始写的时候，我会在每个协程里频繁打印日志，看谁先进入、先结束，慢慢建立起事件循环的心智模型。等你真的理解了协程的挂起和恢复机制，再看 asyncio 生态里的各种库，都会觉得顺理成章。
最后再分享一个小技巧：调试异步代码时，不要只靠 print。可以用调试器给每个 Task 单独打断点，或者用
asyncio.all_tasks()
打印当前所有未完成任务，能很直观地看到有没有协程被卡住。这个习惯能帮你省下大量排查时间，也希望你能早点享受异步带来的效率提升。
