这可能是全网最好懂的 Python 协程教程(附代码)

今天想和大家聊一个让很多 Python 初学者头疼、但又绕不开的话题 —— 协程(Coroutine)

如果你写过爬虫,或者用过 FastAPI、aiohttp 这类框架,一定见过 async def、await 这样的关键字。它们看起来简单,但背后的原理却让不少人一知半解:明明只有一个线程,为什么感觉"同时"在干好几件事?今天我们就把这件事彻底讲清楚。

一、为什么需要协程?

先抛一个问题:如果你要发起 10 个网络请求,每个请求耗时 1 秒,总共需要多久?

用最原始的同步写法,答案是 10 秒——因为程序会老老实实地"发请求 → 等待 → 拿结果",一个接一个来。

import timeimport requests  # 同步请求库deffetch(url):    print(f"开始请求:{url}")    response = requests.get(url)    print(f"完成请求:{url}")return response.status_codestart = time.time()urls = [f"https://httpbin.org/delay/1"for _ in range(10)]for url in urls:    fetch(url)print(f"总耗时:{time.time() - start:.2f} 秒")  # 大约 10 秒

问题很明显:大部分时间程序都在"傻等"网络返回,CPU 其实是闲着的。这种被浪费掉的等待时间,正是协程要解决的核心问题。

解决思路有 3 种:

1. 多进程:开销大,适合 CPU 密集型任务,不适合这种 I/O 等待场景。
2. 多线程:能解决问题,但线程切换有成本,而且 Python 有 GIL(全局解释器锁)限制,并发效果打折扣。
3. 协程:单线程内实现"任务切换",等待 I/O 时主动让出控制权,让其他任务先执行。开销极小,是 I/O 密集型场景的最优解。

二、协程到底是什么?

用一个生活中的例子来理解:

你在厨房同时做 3 道菜。菜 1 需要炖 20 分钟,菜 2 需要焖 15 分钟,菜 3 需要炒 5 分钟。一个聪明的厨师不会傻等菜 1 炖完再做菜 2,而是:启动菜 1 的炖煮 → 趁着这段时间去准备菜 2 → 焖上菜 2 → 再去炒菜 3。

这就是协程的核心思想:遇到"要等待"的操作时,主动把执行权让出去,让别的任务先跑,等结果好了再回来继续。

在 Python 中,协程是通过 async 和 await 这两个关键字,配合 asyncio 这个标准库实现的。

三、核心语法:async 和 await

3.1 async def:定义一个协程函数

用 async def 定义的函数,调用后不会立即执行,而是返回一个 协程对象(coroutine object)

asyncdefsay_hello():    print("你好,协程!")# 直接调用不会执行函数体result = say_hello()print(type(result))  # <class 'coroutine'>

注意看:直接调用 say_hello() 并不会打印"你好,协程!",而是得到一个协程对象。这是初学者最容易踩的第一个坑——协程函数必须被"驱动"起来才会真正执行,单纯调用是没用的。

3.2 await:等待另一个协程完成

await 用来"挂起"当前协程,等待另一个可等待对象(协程、Task 等)执行完成,同时把控制权让给事件循环去做别的事。

import asyncioasyncdefsay_hello():    print("开始打招呼")await asyncio.sleep(1)  # 模拟耗时操作,注意不是 time.sleep    print("你好,协程!")# 用 asyncio.run 来真正驱动协程执行asyncio.run(say_hello())

这里有个关键细节:一定要用 asyncio.sleep() 而不是 time.sleep()。因为 time.sleep() 会让整个线程都阻塞住,别的协程也没法运行;而 asyncio.sleep() 只是"挂起当前协程",事件循环可以趁这段时间去执行别的任务。这是初学者最容易踩的第二个坑。

四、事件循环:协程的调度中心

协程能够"切换执行"的幕后功臣,是事件循环(Event Loop) 你可以把它理解成一个任务调度员,手里拿着一份任务清单,不断循环检查:

• 有没有任务已经"准备好继续执行"了?有就跑一段。
• 有没有任务正在等待 I/O(比如网络请求)?先放一边。
• 所有任务都执行完了吗?执行完就退出循环。

在 Python 3.7 及以后,启动事件循环最简单的方式就是 asyncio.run():

import asyncioasyncdefmain():    print("程序开始")await asyncio.sleep(2)    print("程序结束")asyncio.run(main())  # 创建事件循环,运行 main,结束后自动关闭循环

你不需要手动管理事件循环的创建和关闭,asyncio.run() 会帮你搞定这一切,这也是官方推荐的入口方式。

五、真正的并发:Task 与 gather

前面的例子里,await 是串行等待的——一个协程等完了才轮到下一个。如果想让多个任务真正并发执行,需要用到 asyncio.create_task() 或 asyncio.gather()。

5.1 串行 vs 并发的直观对比

import asyncioimport timeasyncdeftask(name, delay):    print(f"任务 {name} 开始")await asyncio.sleep(delay)    print(f"任务 {name} 完成")returnf"{name} 的结果"# 写法一:串行执行(错误示范,没有利用协程优势)asyncdefrun_serial():    start = time.time()await task("A", 2)await task("B", 2)await task("C", 2)    print(f"串行耗时:{time.time() - start:.2f} 秒")  # 约 6 秒# 写法二:并发执行(正确姿势)asyncdefrun_concurrent():    start = time.time()# gather 会把多个协程"打包"并发运行,等全部完成后统一返回结果    results = await asyncio.gather(        task("A", 2),        task("B", 2),        task("C", 2)    )    print(f"并发耗时:{time.time() - start:.2f} 秒")  # 约 2 秒    print(results)  # ['A 的结果', 'B 的结果', 'C 的结果']asyncio.run(run_concurrent())

可以看到,3 个任务各自耗时 2 秒,串行执行需要 6 秒,而并发执行只需要 2 秒左右——因为 3 个任务的等待时间是"重叠"的,谁也不耽误谁。

5.2 create_task:让任务提前"预约"执行

asyncio.create_task() 会把协程包装成一个 Task 对象,并立即安排它加入事件循环准备执行,而不是等到 await 那一刻才开始排队。

import asyncioasyncdefworker(name, delay):await asyncio.sleep(delay)    print(f"{name} 完成")return nameasyncdefmain():# 创建 Task 后,协程会被立即调度,不需要等到 await    task1 = asyncio.create_task(worker("任务 1", 3))    task2 = asyncio.create_task(worker("任务 2", 1))    print("两个任务已经在后台运行了")# 这里才真正等待结果,但此时任务其实早就开始跑了    result1 = await task1    result2 = await task2    print(f"结果:{result1}, {result2}")asyncio.run(main())

一句话总结 gather 和 create_task 的区别:create_task 负责"启动"一个任务放进后台运行,gather 负责"批量收集"多个协程的结果并等待它们全部完成。实际开发中两者经常搭配使用。

六、实战案例:用 aiohttp 并发抓取多个网页

理论讲了不少,来看一个更贴近实际业务的例子——并发请求多个 URL(注意,这里需要用支持异步的 aiohttp,而不是同步的 requests)。

import asyncioimport aiohttpimport timeasyncdeffetch_url(session, url):"""单个请求的协程函数"""asyncwith session.get(url) as response:# 等待网络返回,这里会自动切换去执行其他协程        data = await response.text()        print(f"{url} 返回长度:{len(data)}")return len(data)asyncdefmain():    urls = ["https://httpbin.org/delay/1","https://httpbin.org/delay/1","https://httpbin.org/delay/1","https://httpbin.org/delay/1","https://httpbin.org/delay/1",    ]    start = time.time()# 用同一个 session 复用连接,性能更好asyncwith aiohttp.ClientSession() as session:# 把 5 个协程一次性交给 gather 并发执行        tasks = [fetch_url(session, url) for url in urls]        results = await asyncio.gather(*tasks)    print(f"5 个请求总耗时:{time.time() - start:.2f} 秒")  # 约 1 秒多,而不是 5 秒asyncio.run(main())

如果每个请求单独耗时 1 秒,串行执行需要 5 秒,而并发执行只需要 1 秒出头。这就是协程在 I/O 密集型场景下的真实威力。

七、新手常踩的 5 个坑

1. 忘记 await:调用协程函数却不加 await,只会得到一个协程对象,函数体根本不会执行。
2. 混用同步阻塞函数:在协程里用了 time.sleep()、requests.get() 这类同步阻塞调用,会把整个事件循环卡死,其他协程全部陪跑。
3. 以为并发数量无限好:如果同时发起成百上千个请求,容易被目标服务器限流或拒绝,建议用 asyncio.Semaphore 控制并发数量。
4. 在普通函数里直接用 await:await 只能出现在 async def 定义的函数内部,否则会直接报语法错误。
5. 异常处理被忽略:asyncio.gather() 默认某个任务出错会导致整体抛异常,可以传入 return_exceptions=True 让它把异常也当作结果返回,不中断其他任务。

补充一个用 Semaphore 控制并发数的实用写法:

import asyncioasyncdeflimited_task(semaphore, name):asyncwith semaphore:  # 获取一个"名额"才能继续执行        print(f"{name} 正在执行")await asyncio.sleep(1)        print(f"{name} 执行完成")asyncdefmain():    semaphore = asyncio.Semaphore(3)  # 最多同时允许 3 个任务并发    tasks = [limited_task(semaphore, f"任务 {i}") for i in range(10)]await asyncio.gather(*tasks)asyncio.run(main())

八、总结

回顾一下今天的核心内容:

• 协程的本质是单线程内的任务切换,遇到 I/O 等待就主动让出执行权,大幅减少无效等待时间。
• async def 用来定义协程函数,调用后返回协程对象,必须被驱动才会执行。
• await 用来挂起当前协程,等待另一个可等待对象完成。
• 事件循环是协程调度的核心,asyncio.run() 是最简单的启动入口。
• create_task 让任务提前进入后台运行,gather 用来并发收集多个协程的结果。
• 协程最适合的场景是 I/O 密集型任务(网络请求、数据库查询、文件读写),而不是 CPU 密集型计算——后者应该用多进程。

协程这个话题第一次接触会有点绕,但只要把"单线程内主动切换任务"这个核心思想理解透了,后面的语法和 API 都是水到渠成的事。建议大家把文中的代码示例跑一遍,亲手体验一下串行和并发的耗时差异,理解会更深刻。

如果这篇文章对你有帮助,欢迎点赞、转发,你的支持是我持续输出的动力!有任何问题也欢迎在评论区交流~