Python 协程：从 yield 到 async/await，异步编程的进化之路
写 Python 做网络请求、爬虫、Web 服务的时候，你有没有遇到过这种情况：程序卡在某个地方等响应，CPU 明明闲着，但就是干等着？
协程就是解决这个问题的。它让你在等待 I/O 的时候去做别的事，而不是傻等。
协程到底是什么？跟线程有什么区别？
先回答一个问题：协程比线程轻量多少？
- 一个线程的内存占用大约 8MB，一个协程大约 2KB
- 线程切换涉及内核态，协程切换只在用户态
- 一台服务器能跑几百个线程，但能跑几万个协程
协程是"协作式"的：它主动让出 CPU（遇到 await 的时候），而不是被操作系统强行打断。这意味着同一时刻只有一个协程在运行，不需要考虑锁的问题。
什么时候用协程？ I/O 密集型任务——网络请求、文件读写、数据库查询。这些操作大部分时间在等外部响应，协程可以利用这些等待时间做别的事。
什么时候别用？ CPU 密集型计算。协程解决不了计算慢的问题，这种场景应该用多进程。
协程的演进：从 yield 到 async/await
Python 的协程经历了几个阶段，了解一下演进脉络，能帮你理解为什么现在是 async/await。
第一阶段：yield（生成器）
Python 2 时代，用生成器来模拟协程。yield 可以暂停函数、返回一个值，下次调用 next() 继续执行。
def generator():
yield 1
yield 2
yield 3
g = generator()
next(g) # 1
next(g) # 2
这已经具备了"暂停-恢复"的能力，但用来做异步调度还太原始。
第二阶段：yield from（委托生成器，Python 3.3）
让一个生成器可以把执行委托给另一个生成器，解决了嵌套调用的问题。
def sub():
yield 1
yield 2
def main():
yield from sub() # 把控制权交给 sub
yield 3
第三阶段：@asyncio.coroutine + yield from（Python 3.4）
Python 3.4 引入了 asyncio 库，用装饰器标记协程。
@asyncio.coroutine
def old_style():
yield from asyncio.sleep(1)
return "done"
这套写法能用，但不够直观——明明是协程，看起来却像生成器。
第四阶段：async/await（Python 3.5+，现在的标准）
这才是你现在应该用的方式。async def 定义协程，await 暂停等待。
async def modern():
await asyncio.sleep(1)
return "done"
一段话总结演进： 从生成器到 yield from 到装饰器到原生语法，async/await 是最终形态——语义清晰、写法直观。写新代码就用这个，前面的了解就行。
async/await 基础用法
定义和运行一个协程：
import asyncio
async def say_hello(name):
print(f"开始：{name}")
await asyncio.sleep(1) # 模拟 I/O 操作
print(f"结束：{name}")
return f"完成：{name}"
# 运行协程
async def main():
result = await say_hello("Alice")
print(result)
asyncio.run(main())
关键点：
- async def 定义协程函数，调用它返回协程对象，不会立即执行
- await 只能在 async def 里用，作用是"暂停当前协程，等异步操作完成"
- asyncio.run() 是 Python 3.7+ 的入口，创建事件循环并运行
能 await 什么？ 协程对象、Task 对象、Future 对象。不能 await 普通函数。
# 错误
await time.sleep(1) # time.sleep 是同步函数
# 正确
await asyncio.sleep(1)
并发执行多个协程
单个协程其实看不出优势，协程真正的威力在于同时执行多个。
gather：并发执行，一起等结果
async def task(name, delay):
await asyncio.sleep(delay)
return f"{name} 完成"
async def main():
results = await asyncio.gather(
task("A", 2),
task("B", 1),
task("C", 1.5),
)
print(results) # ['A 完成', 'B 完成', 'C 完成']
asyncio.run(main())
三个任务总共耗时约 2 秒（最长的那个），而不是 2+1+1.5=4.5 秒。这就是并发的价值。
gather 的结果按传入顺序返回，不是按完成顺序。
create_task：提前创建任务
async def main():
# 任务创建后立即开始执行
task1 = asyncio.create_task(task("A", 2))
task2 = asyncio.create_task(task("B", 1))
# 做点别的事...
# 然后等结果
result1 = await task1
result2 = await task2
create_task 让任务"先跑起来，后面再等结果"，比先收集再 gather 更灵活。
as_completed：谁先完成先处理谁
async def main():
tasks = [task("A", 3), task("B", 1), task("C", 2)]
for coro in asyncio.as_completed(tasks):
result = await coro
print(f"先完成的：{result}")
# 输出顺序：B → C → A
适合"谁先回来先处理谁"的场景，比如爬虫爬多个页面，哪个先回来就先用哪个。
超时控制和并发限制
超时：别让一个慢任务拖死全部
async def main():
try:
result = await asyncio.wait_for(slow_task(), timeout=2)
except asyncio.TimeoutError:
print("超时了，不等了")
wait_for 会在超时后取消任务，适合对响应时间有要求的场景。
并发限制：别一次性发太多请求
sem = asyncio.Semaphore(10) # 最多10个并发
async def fetch_with_limit(session, url):
async with sem: # 占一个名额
async with session.get(url) as resp:
return await resp.text()
爬虫、API 调用时很有用——你不希望一次性发 1000 个请求把服务器打挂。
实战：异步爬虫
用 aiohttp 替代 requests，用协程并发请求。
import asyncio
import aiohttp
import time
urls = [
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/2",
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/3",
]
async def fetch(session, url):
start = time.time()
async with session.get(url) as resp:
content = await resp.text()
elapsed = time.time() - start
return {"url": url, "status": resp.status, "time": elapsed}
async def main():
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks)
for r in results:
print(f"{r['url']} - {r['status']} - {r['time']:.2f}s")
start = time.time()
asyncio.run(main())
print(f"总耗时：{time.time() - start:.2f}s")
4 个请求串行要等 1+2+1+3=7 秒，协程并发只等最长的 3 秒。
异步库 vs 同步库：
用协程的时候，必须用对应的异步库，同步库会阻塞事件循环。
容易踩的坑
坑1：协程里用同步 I/O
# 错误： 阻塞了整个事件循环
async def bad():
time.sleep(1) # 所有协程都卡住
requests.get("https://...") # 同样阻塞
# 正确： 用异步版本
async def good():
await asyncio.sleep(1)
async with aiohttp.ClientSession() as session:
await session.get("https://...")
坑2：忘记 await
async def fetch():
return "data"
async def main():
result = fetch() # 错误 result 是协程对象，不是 "data"
result = await fetch() # 正确 拿到真正的结果
坑3：事件循环重复运行
# 错误： 不行
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.run_until_complete(main()) # RuntimeError
# 正确： 用 asyncio.run()，每次新建循环
asyncio.run(main())
asyncio.run(main())
本文由 AI图鉴自动抓取并翻译，版权归原作者所有；如需查看原文请点下方链接。