Python Asyncio异步编程入门：从协程原理到爬虫并发实战

你是不是也有过这种经历：写了一段爬虫去抓几十个页面，每抓一个都要等上两三秒，如果中途某个接口慢一点，整个程序就卡在那儿干瞪眼。一提python异步编程，很多人第一反应是“得学asyncio？好复杂”。其实拆开看也没那么玄乎——它就是让程序在等待I/O的时候先去做别的事情，回来的时候结果已经准备好了。这篇文章的主角就是Python标准库里的Asyncio，我会从最朴素的场景讲起，把async/await、事件循环、任务调度这些概念用大白话讲清楚，再附上几段可以直接跑的示例代码，以及我用它爬数据和做接口开发时踩过的一些坑。如果你已经会写Python基础语法，准备往爬虫、工具脚本、接口并发这些方向发展，这篇笔记应该能帮你少走不少弯路。

1. 异步到底在解决什么问题

1.1 先看同步代码的痛点

先讲一个大家都会碰到的场景。我们要从三个接口拿数据，如果按顺序一个一个来，代码大概是这样的：

import time

def fetch_data(url):
print(f"开始请求：{url}")
time.sleep(2) # 模拟网络等待
print(f"拿到数据：{url}")
return f"data from {url}"

start = time.time()
data1 = fetch_data("接口A")
data2 = fetch_data("接口B")
data3 = fetch_data("接口C")
print(f"总耗时：{time.time() – start:.2f}s")

实际执行会发现总耗时6秒。这6秒里，程序真正干活的时间可能不到0.01秒，剩下的时间全部花在等待网络响应上。

time.sleep(2)

这个操作是阻塞的，就这么干占着线程，什么事情也做不了。

如果是在爬虫场景里，往往不是3个接口，而是300个、3000个接口。按这种写法，3000个页面串行抓取，假设每个2秒，总耗时就是100分钟起步。这就很难受了。

1.2 为什么说Asyncio适合I/O密集型任务

这里需要先分清楚一个概念：程序等待耗时多，其实分两种类型。一种是输入输出密集型，比如网络请求、文件读写、数据库查询，特点是等待时间远大于计算时间；另一种是计算密集型，比如图片处理、加解密、数值运算，特点是CPU一直在忙。

Asyncio解决的是前一种问题——等待。它做的事情本质上可以理解为：当程序需要等待一个输入输出操作时，先把当前任务挂起来，把控制权交还给事件循环，事件循环立刻调度另一个不需要等待的任务开始执行。等之前那个输入输出完成了，再回来接着往下跑。

我用一个生活例子来类比：同步代码就像一个食堂窗口，打饭的人挨个排在窗口前，前面那个人不拿到饭离开，后面的人就只能一直等着。而Asyncio就像食堂窗口只腾出一个人盯着一排炉灶——他把菜下锅之后，不等它熟，马上转身去给下一个窗口的锅加材料，哪个锅响铃了，就回头处理哪一锅。

这个类比里有几个关键点值得说一下：

• 整个过程是一个线程完成的，不用像多线程那样考虑锁、共享内存互相踩踏的问题；
• 所谓并发，不是CPU同时在算很多件事，而是同一时刻只做一件事，但等待期间可以切去干另一件事；
• 既然是单线程切换，切换成本比线程切换低得多，所以同样的代码用Asyncio往往比用多线程跑网络请求更省资源。

需要提醒的是，如果任务是计算密集型的，比如一个纯循环算一万次大数乘法，Asyncio帮不上忙，反而会因为切换开销变得更慢。这种场景应该考虑多进程或使用C扩展库。

对比一下三种方案的区别，可以放在一个表里：

线程数量

并发原理

适合场景

主要代价

2. Asyncio核心概念拆解

2.1 协程函数：async def 和 await

在Python里，定义一个协程函数只用加一个async关键字：

import asyncio

async def hello():
print("开始执行")
await asyncio.sleep(1)
print("结束执行")

注意一个特别容易踩坑的细节：调用一个协程函数，并不会执行函数体里的任何代码，而是返回一个coroutine对象。

hello() # 这行不会打印任何东西，只返回一个coroutine对象

真正让这个协程跑起来，必须把它放入事件循环，或者用await去等待它。比如：

import asyncio

async def hello():
print("开始执行")
await asyncio.sleep(1)
print("结束执行")

asyncio.run(hello())

这里涉及两个核心关键字：

• async def：声明这是一个协程函数，函数体内部可以包含await表达式；
• await：挂起当前协程，等待另一个可等待对象（awaitable）完成后返回结果，再继续往下执行。

await后面跟的对象，可以是协程、Future、Task，也可以是任何一个实现了

__await__

协议的对象。它的语义很直白：让我等这个东西完成，在等待期间请事件循环去执行别的任务。

第一遍用Asyncio时最容易犯的错误就是：在不能await的地方写await，或者在需要await的地方忘记写await。Python在3.5之后对async/await做了明确区分，如果在普通函数里写await会直接报语法错。

2.2 事件循环与Task：异步的中枢调度器

协程只是一个可被暂停恢复的函数，真正调度它的是事件循环。

事件循环可以理解为一个无限循环的调度器：

在Python 3.7以前，需要自己管理事件循环：

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

Python 3.7之后，官方推荐使用

asyncio.run(main())

一行搞定。这个函数会自己创建事件循环、运行传入的协程、最后关闭循环，省去了一堆样板代码。建议新代码统一用这个写法。

再细一点说，事件循环里跑的最小调度单元其实是Task，而不是协程。Task可以看作协程的包装壳，它把一个协程注册进事件循环，同时记录这个协程的执行状态、结果和异常。

创建Task有两种方式：

task1 = asyncio.create_task(my_coro()) # 推荐，Python 3.7+
task2 = asyncio.ensure_future(my_coro()) # 老写法，仍然可用

asyncio.create_task()

会在当前事件循环里创建一个Task，并将其排入调度队列。注意它必须在事件循环运行期间调用，不能在一个协程还没被asyncio.run包装之前就去创建。

2.3 常用调度API一览

初学者看到Asyncio的API数量可能会有点晕，但实际上日常高频使用的就这几个：

作用

说明

有一个细节值得单独拿出来说：

asyncio.gather

返回的结果列表，顺序永远和传入的可等待对象顺序一致。比如发起10个请求，有的快有的慢，第8个先返回，第3个最后返回，但你拿到结果时下标对应的仍是原来的任务。这对后续数据处理非常方便，不需要自己手动维护任务和结果的映射。

另外，

asyncio.create_task

之后，任务就已经开始调度了，不一定非要等

await

那一刻才跑。所以如果想并发跑多个协程，正确的姿势是：先把所有任务创建出来，然后用

await asyncio.gather(*tasks)

统一等待。如果写成一个任务一个await，那又变成串行了。

3. 三个可直接运行的实操案例

3.1 案例一：理解并发与串行的区别

先看串行写法：

import asyncio, time

async def handle(name):
print(f"{name} 开始处理")
await asyncio.sleep(2)
print(f"{name} 处理完成")

async def main():
start = time.time()
await handle("任务A")
await handle("任务B")
await handle("任务C")
print(f"串行总耗时：{time.time() – start:.2f}s")

asyncio.run(main())

这个输出的总耗时大约是6秒。原因很明确：

await handle("任务A")

会一直等到这个协程结束才返回，然后才执行下一个handle。

再看并发写法：

async def main():
start = time.time()
tasks = [
asyncio.create_task(handle(name))
for name in ["任务A", "任务B", "任务C"]
]
await asyncio.gather(*tasks)
print(f"并发总耗时：{time.time() – start:.2f}s")

asyncio.run(main())

耗时大约是2秒。因为这3个任务都调用了

asyncio.sleep(2)

，相当于告诉事件循环“我先等2秒，这段时间你可以去跑别的任务”，所以事件循环在等待期间分别调度了另外两个协程。三者的2秒等待是重叠的。

这个例子可能显得简单，但它揭示了整个Asyncio最重要的一条规则：并发的收益来自让出控制权。如果某个任务没有那么多次await，没有把控制权交还给事件循环，其他任务再怎么排Task也不会让它提前执行。

3.2 案例二：用gather批量并发

下面把场景换成更实际的接口请求。假设我们要从同一个服务拉取10个页面的数据，用requests库串行写很简单，但在Asyncio里要换思路。

import asyncio
import aiohttp

async def fetch_json(session, url):
async with session.get(url) as resp:
return await resp.json()

async def main():
urls = [f"https://api.example.com/item/{i}" for i in range(1, 11)]
async with aiohttp.ClientSession() as session:
tasks = [asyncio.create_task(fetch_json(session, url)) for url in urls]
results = await asyncio.gather(*tasks)
print(f"共获取 {len(results)} 条数据")

asyncio.run(main())

这里有一个新手很容易踩的坑：如果把

async with aiohttp.ClientSession()

放在每个

fetch_json

里面，每个请求都会新建立一个session，connection pool被浪费，甚至可能直接把目标服务打挂。正确做法是让整个任务组共享一个session，aiohttp的底层连接池会自动复用TCP连接，连接建立的开销被摊薄。

另外注意，这里用

asyncio.gather

而不是逐个

await

。把tasks创建完再统一等待，才能最大化并发；逐个await会变成串行。

3.3 案例三：爬虫实战，并发请求+限流+超时

真实爬虫或者批量接口调用，往往要面对目标服务的限流。一口气开1000个并发请求，很容易被对方拒绝或者被封IP。给网络请求加信号量限制并发数，是必须的一步。

import asyncio
import aiohttp

async def fetch_with_limit(session, url, limit):
async with limit:
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
return resp.status, await resp.text()
except asyncio.TimeoutError:
return 0, "超时"
except Exception as e:
return 0, str(e)

async def main(urls):
max_concurrency = 5
semaphore = asyncio.Semaphore(max_concurrency)
async with aiohttp.ClientSession() as session:
tasks = [asyncio.create_task(fetch_with_limit(session, url, semaphore)) for url in urls]
results = await asyncio.gather(*tasks)
return results

urls = [f"https://example.com/page/{i}" for i in range(1, 21)]
results = asyncio.run(main(urls))

简单解释一下这段代码的设计：

• asyncio.Semaphore(max_concurrency)

  相当于放行令牌，

  async with limit

  进入前先占一个位置，离开时释放；

• aiohttp.ClientTimeout(total=10)

  给每个请求设置了总超时时间，避免一个慢接口把整个程序拖死；

• 异常被捕获后返回状态码0和错误信息，这样即使单个接口失败，整个程序不会崩，后续可以对失败结果统一处理。

运行后，20个页面用5并发，理想情况下分别进行4轮，总耗时大约是单个请求耗时的4倍左右，而不是串行的20倍。如果试了大并发之后发现速度反而没有明显提升，优先检查是不是目标服务有限流策略，同时检查自己的日志里有没有大量超时。适当调低并发数往往比不断调高更有效。

这里还可以再补一个进阶小技巧：如果接口失败需要重试，可以在

fetch_with_limit

外层套循环，比如最多重试3次，间歇用

await asyncio.sleep(1)

。这个sleep是异步的，重试期间其他任务照常推进，不会把整个程序堵死。我一般会再记一份失败列表，最后统一再跑一轮，效果通常比单纯加大并发好得多。

4. 常见问题与排查技巧实录

4.1 RuntimeWarning：协程从未被await

这段可能看起来比较入门，却是很多人的第一次崩溃现场。如果写了这样的代码：

async def get_data():
return 1

def main():
get_data() # 忘写await

Python会在程序结束后弹出一条警告：

RuntimeWarning: coroutine 'get_data' was never awaited

。原因很简单：协程函数返回的是一个coroutine对象，这个对象没有被放入事件循环，也没有被await，所以它一直挂着，等于一个没人管的任务。

解决方式有三种：

• 在async函数内部等待它：

  result = await get_data()

• 在事件循环内创建Task：

  task = asyncio.create_task(get_data())

• 在外层用asyncio.run跑：

  asyncio.run(get_data())

还有一种容易误导人的情况：在

asyncio.gather

里忘记加

await

。所以排查这个警告时，建议先全局搜索

async

定义，再查每一个调用处是否都跟了

await

或

create_task

。

4.2 程序卡住不动：定位事件循环被阻塞

协程程序卡住最常见的原因，不是程序死了，而是事件循环被某个同步阻塞操作卡住了。

回顾一下事件循环的工作方式：它在单线程里不停切换各个协程，而切换的前提是协程执行到

await

时主动让出控制权。如果某个协程里混入了一段同步阻塞代码，比如

time.sleep(2)

、

requests.get(url)

、

subprocess.run()

，事件循环在这几秒内什么事情都干不了，所有其他协程一起被堵在门口。

排查思路可以参考这几步：

• 第一步：看代码里有没有

  time.sleep

  、

  requests

  、

  subprocess.run

  、

  os.system

  这类同步阻塞调用，有就换掉；

• 第二步：把耗时的同步函数用

  await asyncio.to_thread(func, *args)

  丢到线程池去执行，免得污染主循环；

• 第三步：如果是接口调用，给

  asyncio.wait_for

  包一层超时，超过时限强制取消。

举个例子，如果有一个老接口只能通过requests访问，不想换aiohttp，可以这样兼容：

import asyncio, requests

async def get_with_requests(url):
return await asyncio.to_thread(requests.get, url)

async def main():
resp = await get_with_requests("https://example.com")
print(resp.status_code)

asyncio.run(main())

asyncio.to_thread

是Python 3.9提供的能力，内部会用默认的线程池执行器跑同步函数，协程等它的时候，事件循环仍然能调度别的任务，不会全面卡死。这种异步框架里混桥接同步代码的做法，在迁移老代码、对接SDK时非常常用。

4.3 运行环境差异：Jupyter、Windows与Python版本

在不同环境里跑Asyncio，遇到的坑还不太一样。

在Jupyter Notebook里使用

asyncio.run()

，经常会报错：

RuntimeError: asyncio.run() cannot be called from a running event loop

。原因在于IPython从7.0开始默认启动了一个事件循环，你的代码其实是运行在已有循环里的。这个时候不用

asyncio.run

，直接在单元格里写

await main()

即可，相当于让当前协程挂到已有循环上运行。

在Windows平台上，Python 3.8之前的默认事件循环策略和Subprocess配合不好，历史上有段经典设置代码：

asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

不过Python 3.8+在Windows上默认用的是ProactorEventLoop，大部分网络IO场景都很正常，普通入门开发不需要主动改动。真正遇到问题再查这层也不迟。

Python版本方面，建议新项目直接用Python 3.10或3.11以上环境。Asyncio的大部分现代特性，比如

asyncio.run

、

asyncio.create_task

、

asyncio.to_thread

，在3.9/3.10里已经很完整，3.11还引入了

asyncio.TaskGroup

和

asyncio.timeout

，写起来更简洁。如果项目环境锁在3.7以下，有些写法会报错，建议尽早升级。

4.4 开发环境搭建参考

这块给纯新手朋友们一个快速启动参考。如果还没装Python，从官网下载安装包即可，安装时务必勾选“Add Python to PATH”。对Windows用户来说，没有勾选这一项是后面命令行找不到python的最常见原因。

编辑器方面，我推荐VSCode。配置环境时最容易踩的坑就是“明明装了Python，VSCode里pip却装不上包”，往往是右下角选择的解释器不是当前虚拟环境。建议在项目目录下创建虚拟环境：

python -m venv venv
# Windows
venv\\Scripts\\activate
# Linux/macOS
source venv/bin/activate
pip install aiohttp

装好aiohttp之后，在VSCode里输入

python

选择解释器，指向

venv

里那个Python即可。顺手在终端里执行

python -c "import aiohttp; print(aiohttp.__version__)"

验证包是不是装进了正确的环境。这块配置花不了几分钟，但能省后面大量“pip装完却调不到”的困惑。

5. 实操过程中积累的几点心得

最后聊点不太容易在入门文档里看到的东西。

第一，写异步代码前，先在心里把等待点画出来。所谓等待点，就是每个

await

出现的位置。并发不是凭空发生的，只有存在等待点，事件循环才有机会切换过去干别的活。一个纯数字运算的协程，就算外面套一百层async，它也不会和任何任务并发，因为CPU一直在忙，没机会让位。

第二，并发数不是越多越好。我见过一开始用Asyncio的同事，兴冲冲开500个并发抓数据，结果对方服务直接开始返回503。加信号量限流、退避重试，把并发数压到30之后反而速度更快、数据更完整。初期宁可保守一点，先跑通再慢慢调参数。

第三，调试时不要看到一大堆traceback就慌。Asyncio的报错堆栈往往很长，里面层层封装了事件循环和第三方库的内部调用，但99%的问题根源都在你代码里那几行async/await上。优先检查：是不是忘了await、是不是把异步库和同步库混用了、是不是没加超时。学会忽略堆栈中内部的框架帧，直接定位自己的代码行，排查速度会快很多。

如果这篇文章看到这里，你大概已经能跑通最简单的Asyncio并行任务了。再多说一句我的个人体会：Asyncio的学习曲线其实不陡，真正难的是“习惯让出控制权”这种思维转变。我的经验是，先把基础跑顺，再往深走，反而比一上来啃高级特性更高效。