Python asyncio — 事件循环与协程并发模型

💡 一句话理解

asyncio = 一个单线程调度器(事件循环)+ 一群可暂停的任务(协程)。每个任务遇到 await 就说"我在等 IO, 先让别人跑", 控制权交还循环; 循环靠 epoll 同时盯着成千上万个 fd, 谁就绪就让谁继续。单线程换来了无锁、无切换开销、无 GIL 争抢, 代价是: 任何一个人阻塞, 全体卡住。

🧠 必知必会必考 & 必会

协程 vs 线程

线程: OS 抢占式调度, 切换 μs 级, 每个占 MB 级栈; 协程: 用户态协作式调度, 只在 await 处切换, KB 级内存 — 一个"让"字换来万级并发。

async def task():
    await asyncio.sleep(1)          # 关键: 挂起点让出, 单线程跑上千任务
t0 = time.perf_counter()
await asyncio.gather(*[task() for _ in range(1000)])
time.perf_counter() - t0             # → ≈1s — 协作式切换, 每任务 KB 级内存

await 的语义

挂起当前协程 → 控制权交还事件循环 → 等的 Future 完成后被重新排入就绪队列。await 后面必须跟 awaitable(协程/Task/Future)。

result = await coro()          # 关键: 此处挂起让出, 循环去跑别的 Task
await 42                           # → TypeError: int 不是 awaitable
# 合法 awaitable: 协程 / Task / Future / __await__ 实现

create_task 时机

asyncio.create_task(coro()) 立即注册调度; 直接 await coro() 是顺序执行。要并发就先 create_task 再 await — 这是新手最常错的地方。

async def main():
    t = asyncio.create_task(poll_db())   # 关键: 创建即入队调度
    await handle_request()           # poll_db 同时在跑 — 这才是并发
    await t                          # 需要结果时再等

gather vs TaskGroup

gather 一个失败其他照跑(要手动处理); 3.11+ TaskGroup 一个失败自动取消其余并抛 ExceptionGroup — 新代码首选结构化并发。

results = await asyncio.gather(a(), b())   # 一个失败, 另一个照跑
async with asyncio.TaskGroup() as tg:    # 3.11+: 一个失败自动取消其余
    t1 = tg.create_task(a()); t2 = tg.create_task(b())
# TaskGroup 抛 ExceptionGroup, 异常不吞 — 新代码首选

uvloop

事件循环的 libuv 实现(C 层), drop-in 替换, 吞吐通常提升 2~4 倍; uvicorn/gunicorn 生态默认支持。

import uvloop, asyncio
uvloop.install()                     # 关键: drop-in 替换为 libuv, 吞吐 +2~4 倍
asyncio.run(main())
# uvicorn/gunicorn 生态: --loop uvloop 即可

超时与取消

asyncio.wait_for / asyncio.timeout(3.11+): 取消是在下一个 await 点抛 CancelledError — 所以 finally 清理必须写, 且 except CancelledError 要 re-raise。

try:
    await asyncio.wait_for(fetch(), timeout=3)
except asyncio.TimeoutError: retry()
# 关键: 取消 = 在 fetch 的下一个 await 抛 CancelledError
# finally 清理必须写; except CancelledError 要 re-raise

asyncio 的适用边界

高并发 IO(网关、爬虫、代理、聚合接口)是主场; CPU 密集无效(还是那个线程), 低并发内部服务用同步代码反而更简单。

# 主场: 高并发 IO (网关/爬虫/代理) — 万级连接单进程
# 无效: CPU 密集 — 还是那个单线程, 计算期间谁也切不走
await asyncio.to_thread(sync_io)    # 同步 IO 的过渡姿势
await loop.run_in_executor(ProcessPoolExecutor(), cpu_fn)   # CPU 走进程池

🏭 生产实战 real world · 10 场景

场景 1 · 聚合接口并发拉取 N 个下游

商品页聚合 3 个下游, 串行 900ms 并发 300ms — TaskGroup 结构化并发:

async def product_page(sku):
    async with asyncio.TaskGroup() as tg:          # 3.11+ 结构化并发
        t1 = tg.create_task(fetch_price(sku))      # 立即并发, 不是顺序!
        t2 = tg.create_task(fetch_stock(sku))
        t3 = tg.create_task(fetch_reviews(sku))
    return Page(t1.result(), t2.result(), t3.result())
# 任一下游抛错 → 其余自动取消 → 整体失败, 不会半死不活挂着

场景 2 · 必须用同步库时, 扔进线程池

老代码/SDK 只有同步版(pymysql、boto3), 直接调用会卡死循环 — 包一层即可:

async def get_user(uid):
    return await asyncio.to_thread(sync_db.get_user, uid)
# to_thread (3.9+) 默认线程池; CPU 密集则用:
loop.run_in_executor(ProcessPoolExecutor(), heavy_fn, arg)

注意默认线程池上限 min(32, cpu+4) — 慢查询会把池占满, 需要自建池并监控队列。

场景 3 · aiohttp 连接池 + 全局并发闸门

爬虫/网关类服务, 连接复用与并发上限决定稳定性:

connector = aiohttp.TCPConnector(limit=200, limit_per_host=20)  # 全局/单host并发
async with aiohttp.ClientSession(connector=connector,
                          timeout=ClientTimeout(total=3)) as s:
    ...   # Session 必须复用: 每请求新建 Session = 连接不复用 = 性能差一个量级

场景 4 · 优雅停机: 收集并取消在途任务

K8s 发 SIGTERM 后不能丢在途请求: 停接新流量 → 给存量任务 10s 宽限 → 统一取消:

tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
for t in tasks: t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)   # 等清理跑完再退出

场景 5 · asyncio.Queue 生产者消费者背压

有界队列把消费压力自动推回生产端 — 天然限流, 无需额外信号量:

q = asyncio.Queue(maxsize=1000)
async def producer(urls):
    for u in urls:
        await q.put(u)               # 队列满 → 此处挂起, 生产自动减速
    for _ in workers: await q.put(None)   # 毒丸: 通知 worker 退出
async def worker():
    while (u := await q.get()) is not None:
        await crawl(u)
        q.task_done()

场景 6 · 全局并发闸门 Semaphore

下游限"同时在途数"(非速率)时用信号量, 一行包住请求:

sem = asyncio.Semaphore(50)          # 下游最多接受 50 并发
async def call_api(u):
    async with sem:                     # 第 51 个在此排队等待
        return await client.get(u)
results = await asyncio.gather(*[call_api(u) for u in urls])

场景 7 · asyncpg COPY 批量写入

PostgreSQL COPY 协议比逐条 insert 快 20 倍以上, 指标采集入库标配:

import asyncpg
async def bulk_insert(pool, rows):
    async with pool.acquire() as conn:
        await conn.copy_records_to_table(
            "metrics", records=rows,
            columns=["ts", "name", "value"])   # 100 万行秒级

场景 8 · WebSocket 长连接网关

每连接"读协程 + 写队列"两件套, 单写者保证不交错:

async def session(ws):
    outbox = asyncio.Queue()
    async def writer():
        while msg := await outbox.get():   # 唯一写者
            await ws.send(msg)
    w = asyncio.create_task(writer())
    try:
        async for msg in ws:                # 读循环: 万级连接不慌
            await handle(msg, outbox)
    finally:
        w.cancel()                          # 断开: 读循环退出 → 取消写任务

场景 9 · 缓存防击穿(singleflight)

热点 key 过期瞬间, per-key 锁让百个请求只回源一次:

locks: dict[str, asyncio.Lock] = {}
async def get_with_cache(key):
    if (v := cache.get(key)) is not None: return v
    lock = locks.setdefault(key, asyncio.Lock())
    async with lock:                       # 同 key 并发在此排队
        if (v := cache.get(key)) is not None: return v   # 双检: 等到锁时别人已回填
        v = await query_db(key)
        cache.setex(key, 60, v)
        return v

场景 10 · 定时任务与事件循环共存

"每 30s 同步一次配置"不需要调度器, 一个自愈循环任务即可:

async def sync_config_loop(ctx):
    while not ctx.stopping:
        try:                              # 必须包住: 一次异常 = 任务静默死亡
            await reload_config()
        except Exception:
            log.exception("config sync failed, retry next tick")
        await asyncio.sleep(30)
task = asyncio.create_task(sync_config_loop(ctx))   # 保存引用防 GC!

⚠️ 编码注意与常见坑 pitfalls · 20 条

坑 1 · 忘写 await — 裸调用协程函数只是创建了协程对象, 代码根本没执行(只给你一条 RuntimeWarning)。正解: linter 开启 enable=unused-coroutine; review 时盯紧"调用 async 函数却没有 await"。

fetch_user(uid)                     # 错: 只创建协程对象, 函数体一行没执行
# → RuntimeWarning: coroutine 'fetch_user' was never awaited
await fetch_user(uid)               # 对: ruff/flake8 开 unused-coroutine 兜底

坑 2 · 同步阻塞混进 async — requests/time.sleep/pandas 重计算/同步 DB 驱动, 一个就卡死全场, 表现为 P99 毛刺。正解: 全链路异步库; 过渡期 to_thread 包裹。

async def handler():
    r = requests.get(url)           # 错: 卡死全场 — P99 千毫秒毛刺
    time.sleep(0.2)                  # 错: 同罪
    r = await client.get(url)         # 对: 全链路异步库 (aiohttp/httpx)

坑 3 · 顺序 await 当并发 — await a(); await b() 是串行。正解: create_task 后再 await, 或 gather/TaskGroup。

a = await fetch_a(); b = await fetch_b()      # 错: 串行, 耗时 = 和
a, b = await asyncio.gather(fetch_a(), fetch_b())  # 对: 并发, 耗时 ≈ max

坑 4 · CPU 密集用 asyncio — 事件循环还是单线程, 纯计算让所有任务排队。正解: ProcessPoolExecutor / 拆微服务 / 换 Go。

async def bad():
    return sum(i*i for i in range(10**8))   # 错: 计算期间全场排队
res = await loop.run_in_executor(ProcessPoolExecutor(), heavy)  # 对: 进程池

坑 5 · 每请求新建 ClientSession — 连接无法复用, TCP/TLS 握手开销爆炸。正解: Session 应用级单例, 生命周期跟随 app(lifespan)。

async def handler(url):                      # 错: 每请求新建, TCP/TLS 握手开销爆炸
    async with aiohttp.ClientSession() as s: ...
app.state.http = aiohttp.ClientSession(...)   # 对: 应用级单例, lifespan 管生命周期

坑 6 · 取消时吞异常 — except Exception 会把 CancelledError(继承 BaseException, 3.8+)之外的取消链弄乱, 任务"杀不死"。正解: except CancelledError: 清理后 raise; 不要 bare except。

try: await work()
except: log("err")                          # 错: bare except 连 CancelledError 一起吞, 杀不死
except CancelledError: cleanup(); raise  # 对: 清理后 re-raise 让取消传播

坑 7 · 跨线程碰事件循环 — 其他线程不能直接操作 loop 内对象。正解: loop.call_soon_threadsafe 或 asyncio.run_coroutine_threadsafe。

threading.Thread(target=lambda: loop.run_until_complete(coro()))  # 错: 循环已在别的线程跑
asyncio.run_coroutine_threadsafe(coro(), loop).result()   # 对: 线程安全提交
loop.call_soon_threadsafe(cb)                             # 对: 只排回调

坑 8 · 把协程对象当返回值用 — result = fetch() 没加 await, result 是 coroutine 对象, 后续 result.json() 报 AttributeError。正解: 看到 "never awaited" 警告立刻回头补 await。

result = fetch_json(url)            # 错: result 是 coroutine 对象, 不是数据
result["id"]                        # → TypeError: 'coroutine' object is not subscriptable
result = await fetch_json(url)        # 对: 见 "never awaited" 警告立刻补 await

坑 9 · 忘了 async with / async for — 异步上下文管理器用普通 with 打开, __aenter__ 未被等待, 资源未就绪。正解: aiofiles/aiohttp 全家必须 async with; lint 开 async 检查。

f = aiofiles.open(path)            # 错: 返回协程, 普通 with 打不开异步资源
async with aiofiles.open(path) as f: ...  # 对: __aenter__ 被 await, 就绪后才用
async for msg in ws: ...             # 异步迭代器同理用 async for

坑 10 · 事件循环嵌套 — 在已运行的循环里再调 asyncio.run() → RuntimeError。正解: asyncio.run 只放最外层入口; 库代码提供 async API 由调用方驱动。

async def handler():
    asyncio.run(sub())              # 错: → RuntimeError: asyncio.run() cannot be called from a running event loop
    await sub()                      # 对: run 只放最外层入口, 库代码提供 async API

坑 11 · gather(return_exceptions=True) 吞错 — 返回值里混着异常对象, 不检查就当正常数据用。正解: 消费前逐个 isinstance 校验; 或改用 TaskGroup 让失败显式冒泡。

res = await asyncio.gather(*tasks, return_exceptions=True)
rows = [r["data"] for r in res]   # 错: r 可能是 Exception 对象, 当数据必炸
rows = [r["data"] for r in res if not isinstance(r, BaseException)]  # 对: 逐个校验

坑 12 · asyncio.wait 超时后不取消 — wait 返回的 pending 任务不会被自动取消, 悄悄泄漏。正解: 超时后显式 cancel pending; 新代码优先 wait_for/TaskGroup。

done, pending = await asyncio.wait(tasks, timeout=5)
return [t.result() for t in done]   # 错: pending 悄悄泄漏, 还在后台跑
for t in pending: t.cancel()      # 对: 超时后显式取消; 新代码用 wait_for/TaskGroup

坑 13 · fire-and-forget 任务被 GC — create_task(coro()) 不保存引用, 官方文档明确任务可能被垃圾回收中途消失。正解: 收进集合, 完成回调里移除; 或 TaskGroup 管理。

asyncio.create_task(sync_loop())    # 错: 不存引用 → 官方文档明确可能被 GC 中途消失
bg = set(); t = asyncio.create_task(sync_loop()); bg.add(t)
t.add_done_callback(bg.discard)     # 对: 收进集合, 完成回调里移除; 或 TaskGroup

坑 14 · finally 里的同步阻塞 — 主路径全异步, finally 里一个同步 close(慢 socket/文件)照样卡全循环。正解: 清理也用异步接口, 同步残留包 to_thread。

try: await serve()
finally: sock.close()               # 错: 同步 close 慢 socket, 照样卡全循环
finally: await asyncio.to_thread(sock.close)   # 对: 清理异步化

坑 15 · 长计算不让出 — 纯 CPU 循环中间没有 await, 其他任务全部排队。正解: 分块间 await asyncio.sleep(0) 让出; 本质解法是进程池。

while data := feed():               # 错: 纯 CPU 循环无 await, 其他任务全部排队
    crunch(data)
for chunk in chunks:
    crunch(chunk); await asyncio.sleep(0)  # 对: 分块间让出; 本质解法是进程池

坑 16 · 定时任务累计漂移 — while True: work(); await sleep(60) 实际周期 = 60 + work 耗时, 一天漂几分钟。正解: 记录下次目标时刻, sleep(max(0, next - now))。

while True:                          # 错: 实际周期 = 60s + work 耗时, 越跑越漂
    work(); await asyncio.sleep(60)
nxt = time.monotonic() + 60        # 对: 记目标时刻, 睡到点
while True: work(); nxt += 60; await asyncio.sleep(max(0, nxt - time.monotonic()))

坑 17 · 多任务并发写同一文件 — asyncio 不解决 IO 交错, 并发 append 同一文件照样错行。正解: 单写协程 + 队列(或 logging QueueHandler); 文件用 aiofiles。

await asyncio.gather(*[append_log(l) for l in lines])  # 错: IO 交错照样错行
q = asyncio.Queue()                 # 对: 单写协程 + 队列, 顺序天然保证
async def writer(): ...             # 唯一写者; 文件用 aiofiles

坑 18 · __aexit__ 异常路径未测 — 正常路径通了, 异常路径的清理 await 抛错把原始异常盖掉。正解: 单测覆盖异常分支; __aexit__ 内部再 try/except 保护。

async def __aexit__(self, *exc):
    await self.close()              # 错: 异常路径里 close 再抛错 → 盖掉原始异常
# 对: 清理内部自带保护 + 单测覆盖异常分支
try: await self.close()
except Exception: log.exception("cleanup fail")

坑 19 · uvloop 兼容性想当然 — 个别库(自定义 DNS/monkey patch)在 uvloop 下行为不同。正解: 切换前跑全量集成测试, 保留环境变量一键回退。

# 错: 直接切 uvloop 上生产 — 自定义 DNS/monkey patch 行为可能不同
# 对: 切换前跑全量集成测试, 保留一键回退
UVLOOP=0 python -m app            # 环境变量开关, 出问题秒回默认循环

坑 20 · Ctrl+C 后清理没跑 — KeyboardInterrupt 直接冒到顶层, 取消传播混乱、任务残留。正解: 顶层捕获后显式 cancel + gather(见场景 4), 再退出循环。

# 错: KeyboardInterrupt 直接冒到顶层, 在途任务残留
try: asyncio.run(main())
except KeyboardInterrupt: pass    # 一带而过, 清理没跑
# 对: 顶层捕获后显式 cancel + gather 等清理跑完再退出 (见优雅停机场景)

Python · asyncio 事件循环

协程与 Task

事件循环 = 单线程调度器

阻塞是唯一的罪

💡 一句话理解

🧠 必知必会必考 & 必会

🏭 生产实战 real world · 10 场景

场景 1 · 聚合接口并发拉取 N 个下游

场景 2 · 必须用同步库时, 扔进线程池

场景 3 · aiohttp 连接池 + 全局并发闸门

场景 4 · 优雅停机: 收集并取消在途任务

场景 5 · asyncio.Queue 生产者消费者背压

场景 6 · 全局并发闸门 Semaphore

场景 7 · asyncpg COPY 批量写入

场景 8 · WebSocket 长连接网关

场景 9 · 缓存防击穿(singleflight)

场景 10 · 定时任务与事件循环共存

⚠️ 编码注意与常见坑 pitfalls · 20 条

Python · asyncio 事件循环

协程与 Task

事件循环 = 单线程调度器

阻塞是唯一的罪

💡 一句话理解

🧠 必知必会 必考 & 必会

🏭 生产实战 real world · 10 场景

场景 1 · 聚合接口并发拉取 N 个下游

场景 2 · 必须用同步库时, 扔进线程池

场景 3 · aiohttp 连接池 + 全局并发闸门

场景 4 · 优雅停机: 收集并取消在途任务

场景 5 · asyncio.Queue 生产者消费者背压

场景 6 · 全局并发闸门 Semaphore

场景 7 · asyncpg COPY 批量写入

场景 8 · WebSocket 长连接网关

场景 9 · 缓存防击穿(singleflight)

场景 10 · 定时任务与事件循环共存

⚠️ 编码注意与常见坑 pitfalls · 20 条

🧠 必知必会必考 & 必会