Python asyncio 异步编程实战：用协程与事件循环构建高性能网络服务

导语你有没有写过这样的程序单进程里依次请求 100 个接口前面一个没回来后面 99 个只能干等着或者 FastAPI 路由明明标了async def压测一上来 QPS 却怎么也上不去。问题往往不在机器而在于你把等待当成了占用。声明本文基于个人使用体验非商业推广。摘要asyncio 用事件循环和协程把 IO 等待转化为可以切换的间隙让单进程也能撑起海量并发。本文从底层模型讲起结合 aiohttp 并发请求与 FastAPI 异步接口两个实战带你落地一套可复现的高性能网络服务写法。一、为什么需要 asyncio从同步阻塞到协程并发先说一个最朴素的痛点。下面这段同步代码请求 3 个接口总耗时是三个请求之和import requests, time def sync_fetch(urls): start time.perf_counter() for u in urls: requests.get(u) # 阻塞当前线程在这里卡住啥也干不了 print(f耗时 {time.perf_counter() - start:.2f}s)requests.get是阻塞调用线程发完请求后就停在recv上等服务器回包CPU 空转。如果三个请求各耗时 1 秒整体就是 3 秒。这种同步阻塞模型在 IO 密集型场景网络、磁盘、数据库下CPU 利用率极低。选型上先分清两类负载IO 密集型程序大部分时间在等网络/磁盘真正算的时间很短。爬虫、网关、聚合接口都属于这类是 asyncio 的主场。CPU 密集型大量计算图像处理、加密、科学计算。这类用多进程multiprocessing才能真正并行asyncio 帮不上忙反而会因单线程被长计算占满而让出不了事件循环。asyncio 的核心思路很直白单进程里跑一个调度器谁在等 IO 谁就先让出来让别的协程先跑。于是一个进程在等待 A 接口回包时可以去处理 B 接口的请求表面上同时服务了海量连接。这就是协程并发。二、核心模型事件循环、协程与 Task理解 asyncio 只需要抓住三个名词。事件循环Event Loop可以理解成一位乐队指挥。它手里有一张任务清单不断问现在谁在等 IO谁可以跑了哪个协程遇到await主动让出指挥就转去调度下一个准备好的协程。整个程序始终跑在同一个线程里靠切换而非多线程来实现并发。协程Coroutine是用async def定义的函数。关键认知async def定义出来的对象仅创建不执行。下面这行只是造了一个协程对象它一句话都不会跑async def hello(): print(hi) coro hello() # 只是对象没有打印任何东西要让它跑必须被事件循环调度。最外层的入口由asyncio.run()负责——它帮你创建事件循环、运行协程、最后清理import asyncio async def say_after(delay: float, msg: str) - str: await asyncio.sleep(delay) # await 时把控制权交还事件循环 return msg async def main() - None: # create_task 立即把协程交给事件循环调度不等它跑完 t1 asyncio.create_task(say_after(1.0, hello)) t2 asyncio.create_task(say_after(1.0, world)) # gather 并发等待多个可等待对象按传入顺序返回结果列表 results await asyncio.gather(t1, t2) print(results) # [hello, world] if __name__ __main__: asyncio.run(main()) # 创建/运行/销毁事件循环只调用一次asyncio.run()与asyncio.create_task()的分工要分清run是程序的唯一入口负责生命周期create_task是在已有循环内把一个协程登记成待调度的任务让它真正并发起来。最后记住能被await的东西只有三类统称可等待对象Awaitable协程coroutineasync def函数返回的对象。Task协程被create_task包装后的调度单元是正在跑的协程。Future更底层的未来结果占位符Task 本身继承自 Future日常写业务基本碰不到但理解它有助于读懂源码。三、并发控制asyncio.gather 与任务编排并发的扇出主要靠asyncio.gather。它接收一个或多个可等待对象并发运行并聚合结果import asyncio async def main(): r await asyncio.gather( say_after(1, a), say_after(2, b), say_after(1, c), ) print(r) # [a, b, c] asyncio.run(main())三个协程并发总耗时约等于最慢那一个约 2 秒而不是 1214 秒。这就是并发批量调用的收益。生产环境一定要加return_exceptionsTrue。默认情况下gather 里只要有一个协程抛异常整个 gather 会立刻把异常抛出来其它已成功的结果也拿不到。加上这个参数后异常会被当作普通结果返回保留在对应位置整体不会被单个失败打断import asyncio async def boom(): raise ValueError(挂了) async def main(): r await asyncio.gather(boom(), asyncio.sleep(0, resultok), return_exceptionsTrue) print(r) # [ValueError(挂了), ok] —— 各自结果都在 asyncio.run(main())拿到结果后记得逐个判断是不是Exception类型再做分支处理而不是无脑当正常数据用。四、实战一用 aiohttp 构建并发网络请求标准库asyncio本身不提供 HTTP 客户端实战里常用aiohttp。它的ClientSession是一个异步上下文管理器推荐用async with保证连接正确关闭。核心模式是封装一个fetch协程再用gather并发抓多个 URLimport asyncio import aiohttp async def fetch_url(session: aiohttp.ClientSession, url: str) - str: # ClientTimeout 限定整体耗时防止单个慢请求拖垮整个并发池 timeout aiohttp.ClientTimeout(total10) try: async with session.get(url, timeouttimeout) as resp: resp.raise_for_status() # 4xx/5xx 会抛 ClientResponseError return await resp.text() except aiohttp.ClientError as e: return fERROR:{url}:{e} # 把异常转成字符串便于后续统计 async def main() - None: urls [ https://httpbin.org/delay/1, https://httpbin.org/delay/2, https://httpbin.org/status/404, ] async with aiohttp.ClientSession() as session: # 用生成器表达式批量造协程return_exceptions 让个别失败不影响整体 pages await asyncio.gather( *(fetch_url(session, u) for u in urls), return_exceptionsTrue, ) for url, html in zip(urls, pages): print(url, -, len(html)) if __name__ __main__: asyncio.run(main())三个要点复用同一个ClientSession。它内部维护连接池每次请求都新建 session 既慢又容易耗尽资源。超时必须显式设置。ClientTimeout(total10)是整体上限避免某个 URL 卡死导致整个 gather 一直挂着。异常就地消化。把ClientError转成带标记的字符串返回配合return_exceptionsTrue批量抓取时不会因为一个 404 就让全部结果丢失。五、实战二FastAPI 异步接口构建高性能服务到了服务层async def路由是 asyncio 发挥作用的地方。当路由里出现await比如等数据库、等下游 HTTP事件循环会在等待期间去服务其它请求于是单进程也能扛高并发 IO。下面这个聚合接口并发拉取两个外部用户服务再合并返回import asyncio import aiohttp from contextlib import asynccontextmanager from fastapi import FastAPI async def fetch_user(uid: int) - dict: async with aiohttp.ClientSession() as session: async with session.get(fhttps://api.example.com/users/{uid}) as resp: return await resp.json() asynccontextmanager async def lifespan(app: FastAPI): # 启动阶段创建后台任务交给事件循环调度 app.state.bg asyncio.create_task(periodic_sync()) yield app.state.bg.cancel() # 关闭阶段取消任务避免协程泄漏 async def periodic_sync() - None: while True: await asyncio.sleep(60) # 实际的增量同步逻辑…… app FastAPI(lifespanlifespan) app.get(/aggregate) async def aggregate(): # await 处让出事件循环两个外部调用并行而非串行 u1, u2 await asyncio.gather(fetch_user(1), fetch_user(2)) return {u1: u1, u2: u2}这里有几个工程上容易出错的点多外部 API 聚合用gather并行。若写成两次顺序await耗时就是两者相加用gather则约等于最慢一个。异步后台任务用asyncio.create_task。上例用lifespan在启动期登记后台任务注意on_event(startup)已废弃新项目建议用lifespan。关闭时一定要cancel()否则协程会泄漏。避坑红线协程内严禁调用阻塞操作。绝不要在async def里写requests.get(...)或time.sleep(...)它们会卡死整个事件循环让所有并发请求一起 stalled。CPU 密集活儿请用run_in_executor卸载到线程池见下一节。六、调试与性能最佳实践写异步代码最怕看起来在跑其实全卡住。几条实用手段开启 asyncio 调试模式让循环检测超过 100ms 的慢回调并告警帮你发现偷偷阻塞的事件循环import asyncio async def main(): ... asyncio.run(main(), debugTrue) # 或用环境变量 PYTHONASYNCIODEBUG1run_in_executor卸载阻塞调用。如果必须调用同步库如某些不支持异步的 SDK把它丢到默认线程池别堵在主线程import asyncio, time def blocking_work(x): time.sleep(2) # 假设这是改不了源的同步阻塞库 return x * 2 async def main(): loop asyncio.get_running_loop() # run_in_executor 把阻塞函数交给线程池await 处主循环仍可调度其它协程 result await loop.run_in_executor(None, blocking_work, 21) print(result) # 42 asyncio.run(main())三条黄金法则务必刻进肌肉记忆await不可省略。写了asyncio.sleep(1)却没await协程不会等待逻辑直接跳过。同步函数里不能直接await。await只能出现在async def内部普通函数里写会语法报错。异步上下文里不混阻塞调用。一旦在协程里写了requests/time.sleep整个事件循环都会被打断并发优势归零。总结asyncio 不是银弹但它是 Python 写 IO 密集型网络服务的标配。回想本文主线事件循环是调度核心async/await让出控制权create_task/gather负责并发编排aiohttp 解决并发请求FastAPI 的async def把并发优势带到线上服务。踩坑大多集中在协程里混了阻塞调用和忘了await这两类debug 模式和run_in_executor能兜住大部分性能问题。如果你也在用 asyncio 搭服务欢迎在评论区聊聊踩过的坑。觉得有用可以收藏下次写异步接口时照着第五、六节直接套。参考来源Python 官方asyncio 概念概览 https://docs.python.org/zh-cn/3/howto/a-conceptual-overview-of-asyncio.htmlPython 官方asyncio 协程与任务 https://docs.python.org/3/library/asyncio-task.htmlCSDN 站内相关asyncio 并发实战解析 https://blog.csdn.net/SearchB/article/details/161058580© 2026 | 转载请注明出处