3 行代码把 asyncio 吞吐拉高 2–4 倍：uvloop.run() 与 Runner loop_factory 实战

3 行代码把 asyncio 吞吐拉高 2–4 倍：uvloop.run() 与 Runner loop_factory 实战

【免费下载链接】uvloopUltra fast asyncio event loop.项目地址: https://gitcode.com/gh_mirrors/uv/uvloop

你的 Python 异步服务跑了半年，业务代码反复优化，吞吐还是上不去。问题可能不在代码——事件循环本身就是天花板。uvloop 用uvloop.run()这个函数，靠一个loop_factory参数挂进标准库 Runner 接口，把 asyncio 吞吐拉高 2–4 倍。而你要改的，只有入口那一行。

瓶颈不在业务代码：echo server 的吞吐被事件循环卡住

先甩结论：官方口径是 uvloop 让 asyncio 快 2–4 倍。这个数字来自它自己的基准——同一个 echo server，三种收发路径、不同报文大小，横轴拉满对比两条事件循环。

三条路径分别对应三种写法：

基准脚本在 examples/bench/，echoserver.py是服务端，echoclient.py是压测端，--msize调报文大小、--num调消息总数（默认 200000 条）。

图里真正该记住的只有一点：差距不挑写法。最裸的sock_recv/sock_sendall也有明显差值，说明天花板在事件循环，不在你的 handler。

3 行代码完成替换：uvloop.run() 的 Before/After 对照

从 uvloop 0.18 起，推荐入口就是uvloop.run()。项目改造量是一行。

改之前（标准 asyncio）：

import asyncio async def main(): ... asyncio.run(main())

改之后（uvloop）：

import uvloop async def main(): ... uvloop.run(main())

uvloop.run()的签名是run(main, *, loop_factory=new_event_loop, debug=None, **run_kwargs)。debug等参数原样透传给底层的asyncio.run()，所以uvloop.run(main(), debug=True)的用法和asyncio.run(main(), debug=True)完全对得上，不用改业务。

uvloop.run() 内部按版本分了三叉

看 uvloop/init.py，run()按 Python 版本分成三支：

• ≤3.10：手动set_event_loop(loop)+run_until_complete，finally里取消所有任务、关 async 生成器、关循环；
• 3.11：with asyncio.Runner(loop_factory=loop_factory, debug=debug) as runner: runner.run(wrapper())；
• ≥3.12：直接asyncio.run(wrapper(), loop_factory=loop_factory, debug=debug, **run_kwargs)。

三支写法不同，收敛点只有一个：把uvloop.new_event_loop当作loop_factory塞进标准库的 Runner / asyncio.run()。这就是"3 行代码"的全部秘密——uvloop 不重写事件循环生命周期，只替换产出循环的那个工厂。

loop_factory 参数：Runner 接口真正的接合点

Python 3.11 给asyncio.Runner暴露了loop_factory，这正是 uvloop 瞄的接口。README 里 3.11+ 的直写法是：

import asyncio import uvloop with asyncio.Runner(loop_factory=uvloop.new_event_loop) as runner: runner.run(main())

它和uvloop.run()等价。后者多了一层护栏：wrapper()里用asyncio._get_running_loop()拿到运行中的循环，isinstance(loop, Loop)校验，不是 uvloop 的循环就抛TypeError: uvloop.run() uses a non-uvloop event loop。

测试 tests/test_runner.py 正是钉死这条路径：传入一个非 uvloop 的工厂（asyncio.DefaultEventLoopPolicy().new_event_loop）必须抛TypeError；正常路径则断言isinstance(loop, uvloop.Loop)且debug=True时loop.get_debug()为真。所以 3.11+ 上有两种等价写法，uvloop.run()是带校验的语法糖。

什么时候该上 uvloop，什么时候别上：一张决策表

uvloop 定位是 asyncio 事件循环的drop-in 替换——官方开发文档明确"任何偏离参考事件循环的行为都算 bug"。多数场景即插即用，但边界得看清：

该上：IO 密集的网络服务——代理、网关、高并发长连接，事件循环在热路径上，收益最大；你已在 3.11+，loop_factory是一等参数，接合成本最低。

该犹豫：Python <3.11 时uvloop.run()走手动管理分支（set_event_loop、任务取消都自己兜底），若项目里有大量嵌套循环或手动管理循环的场景，先回归验证；还在用install()/EventLoopPolicy的话，升级新版到 3.12+ 会吃到DeprecationWarning，顺手迁到uvloop.run()。

一步验证替换是否生效：debug=True 加 isinstance 断言

最稳的验证是看运行时循环类型。在main()里打一行：

import asyncio, uvloop async def main(): loop = asyncio.get_running_loop() print(type(loop).__name__) # 应为 uvloop 的 Loop

官方测试套件用的就是这个断言：

async def main(): loop = asyncio.get_running_loop() assert isinstance(loop, uvloop.Loop) assert loop.get_debug()

要开调试就uvloop.run(main(), debug=True)，debug会透传成loop.set_debug(debug)。想在 uvloop 源码树里跑全量回归，make test会把全部单测跑两遍，一遍带PYTHONASYNCIODEBUG=1、一遍不带。

下一步行动清单

1. 安装：先pip install -U pip再pip install uvloop，老版 pip 装 C 扩展容易翻车。
2. 替换：把入口asyncio.run(main())改成uvloop.run(main())，业务代码一行不动。
3. 验证：在main()里断言isinstance(asyncio.get_running_loop(), uvloop.Loop)，确认换的是 uvloop 循环。
4. 压测：用examples/bench/的echoserver.py+echoclient.py，固定--msize、--num对比替换前后吞吐。
5. 回归：跑项目自身测试；uvloop 源码树直接make test，会带/不带异步调试各跑一遍。

延伸阅读

• docs/api/：uvloop.Loop、new_event_loop、EventLoopPolicy的完整 API 参考，查具体方法行为时翻这里。
• examples/bench/：echo 基准的完整客户端与服务端脚本，照它复现 2–4 倍差距、校准自己的部署。

【免费下载链接】uvloopUltra fast asyncio event loop.项目地址: https://gitcode.com/gh_mirrors/uv/uvloop