Python asyncio 模块
Coroutines, the event loop, gather vs create_task, timeouts and the async/await mental model.
Key PointsAsync gives concurrency on a single thread: await yields control while waiting, so never call blocking code inside a coroutine.
Mental Model
async def defines a coroutine; calling it returns an awaitable that does nothing until awaited or scheduled. One event loop drives many coroutines, switching at each await.
Python
import asyncio
async def fetch(name, delay):
print("start", name)
await asyncio.sleep(delay) # 让出控制权，不阻塞线程
print("end", name)
return name
async def main():
await fetch("a", 1) # 串行：总耗时约 1s
await asyncio.gather(fetch("b", 1), fetch("c", 1)) # 并发：总耗时约 1s
asyncio.run(main())gather vs create_task
| Construct | Behaviour | Failure mode |
|---|---|---|
| await f() | Runs inline, serial | Propagates immediately |
| asyncio.gather(a, b) | Run concurrently, collect results | First error propagates, others keep running |
| asyncio.create_task(c) | Schedule now, await later | Un-awaited exceptions are logged |
| asyncio.TaskGroup() | 3.11+ structured concurrency | Cancels siblings on failure |
Python
async def main():
# 并发 + 结果按传入顺序返回
a, b = await asyncio.gather(fetch("x", 1), fetch("y", 2))
# 先调度，后等待（可以中间做别的事）
task = asyncio.create_task(fetch("z", 1))
print("doing something else")
result = await task
# 3.11+：任一失败则整组取消
async with asyncio.TaskGroup() as tg:
tg.create_task(fetch("p", 1))
tg.create_task(fetch("q", 1))Timeouts
Python
async def main():
try:
async with asyncio.timeout(2): # 3.11+
await fetch("slow", 5)
except TimeoutError:
print("timed out")
# 3.8+ 兼容写法
# await asyncio.wait_for(fetch("slow", 5), timeout=2)Do Not Block the Loop
Python
import asyncio
import time
async def bad():
time.sleep(2) # 阻塞整个事件循环，所有协程停摆
async def good():
await asyncio.sleep(2) # 让出控制权
# 必须调同步库时，丢到线程池
async def run_blocking():
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, time.sleep, 2)A single blocking call (requests, time.sleep, heavy CPU) inside a coroutine serialises your entire service. Use an async client (httpx/aiohttp) or run_in_executor.
心智模型
async def 定义协程；直接调用只会返回可等待对象，不执行任何代码，直到被 await 或被调度。一个事件循环驱动多个协程，在每个 await 处切换。
Python
import asyncio
async def fetch(name, delay):
print("start", name)
await asyncio.sleep(delay) # 让出控制权，不阻塞线程
print("end", name)
return name
async def main():
await fetch("a", 1) # 串行：总耗时约 1s
await asyncio.gather(fetch("b", 1), fetch("c", 1)) # 并发：总耗时约 1s
asyncio.run(main())gather 与 create_task
| 写法 | 行为 | 异常行为 |
|---|---|---|
| await f() | 内联执行，串行 | 立即抛出 |
| asyncio.gather(a, b) | 并发执行并收集结果 | 首个异常抛出，其余继续 |
| asyncio.create_task(c) | 立即调度，稍后 await | 未 await 的异常只记日志 |
| asyncio.TaskGroup() | 3.11+ 结构化并发 | 失败时取消兄弟任务 |
Python
async def main():
# 并发 + 结果按传入顺序返回
a, b = await asyncio.gather(fetch("x", 1), fetch("y", 2))
# 先调度，后等待（可以中间做别的事）
task = asyncio.create_task(fetch("z", 1))
print("doing something else")
result = await task
# 3.11+：任一失败则整组取消
async with asyncio.TaskGroup() as tg:
tg.create_task(fetch("p", 1))
tg.create_task(fetch("q", 1))超时
Python
async def main():
try:
async with asyncio.timeout(2): # 3.11+
await fetch("slow", 5)
except TimeoutError:
print("timed out")
# 3.8+ 兼容写法
# await asyncio.wait_for(fetch("slow", 5), timeout=2)不要阻塞事件循环
Python
import asyncio
import time
async def bad():
time.sleep(2) # 阻塞整个事件循环，所有协程停摆
async def good():
await asyncio.sleep(2) # 让出控制权
# 必须调同步库时，丢到线程池
async def run_blocking():
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, time.sleep, 2)协程里一次阻塞调用（requests、time.sleep、重 CPU 计算）就会把整个服务串行化。改用异步客户端（httpx/aiohttp）或 run_in_executor。