Python3 协程：异步编程的核心机制与实战

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 同步库：

场景	同步库	异步库
HTTP 请求	requests	aiohttp
文件读写	open()	aiofiles
数据库	psycopg2	asyncpg / aiomysql

用协程的时候，必须用对应的异步库，同步库会阻塞事件循环。

容易踩的坑

坑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())