Python asyncio.eager_task_factory 怎么用：缓存命中提速与任务顺序回归检查

线上接口缓存命中率很高的场景下，asyncio.create_task() 仍要把协程交给事件循环安排一次调度，这点额外开销搞不好比真正的业务代码耗时还突出。Python 的 asyncio.eager_task_factory 提供了更高效的实现路径：协程创建后先同步执行，只有遇到阻塞操作才回到事件循环。

先确认这项改动改变了什么

asyncio.eager_task_factory 不是一个只影响性能数字的提速开关。设置到事件循环后，协程在 Task 构造时就会立刻开始执行；如果它直接返回或抛出异常，就不会再排进事件循环队列。如果执行到 await 后发生阻塞，任务才会交给事件循环继续推进。

这个语义在 Python 3.12 已经加入标准库。Python 3.14 又让 asyncio.create_task()、TaskGroup.create_task() 透传更多关键字参数，迁移时要把“全局工厂配置”和“单次任务创建”的逻辑分开梳理。下面的示例用 Python 3.14 写法实现，也会标出适配旧版本的兼容边界。

旧写法与新工厂的差异

用一个缓存协程验证同步完成

先准备两类协程：read_cache() 命中时直接返回，未命中时执行一次真正的异步等待。这样测试结果不会把“减少调度开销”和“网络请求变快”两个因素混在一起。

import asyncio
import time

async def read_cache(key: str):
    if key == "hit":
        return {"key": key, "source": "memory"}
    await asyncio.sleep(0.002)
    return {"key": key, "source": "remote"}

async def measure(factory=None):
    loop = asyncio.get_running_loop()
    old_factory = loop.get_task_factory()
    if factory is not None:
        loop.set_task_factory(factory)
    try:
        started = time.perf_counter()
        tasks = [asyncio.create_task(read_cache("hit")) for _ in range(1000)]
        values = await asyncio.gather(*tasks)
        elapsed = time.perf_counter() - started
        return elapsed, values[0]
    finally:
        loop.set_task_factory(old_factory)

async def main():
    normal = await measure()
    eager = await measure(asyncio.eager_task_factory)
    print("normal:", normal[0], normal[1])
    print("eager :", eager[0], eager[1])

asyncio.run(main())

这段实验只比较同一进程、同一批输入下的相对性能变化，不会把单次运行的毫秒数当成性能承诺。实际项目里最好预热解释器，重复多轮测试，分别记录命中和未命中场景的耗时分位数，最后确认事件循环没有残留的未处理任务。

迁移时最容易漏掉的三个风险

不要用创建顺序推断业务顺序

开启 eager 工厂后，第一项命中缓存的协程可能在第二项任务创建之前就已经执行完成。如果代码靠“任务创建的先后”维护列表、写日志或者更新状态，就要改成用显式序号、结果排序或者结构化并发的方式处理，不能依赖事件循环的默认调度行为。

同步异常可能更早暴露

一个没有任何阻塞点的协程可以在任务创建期间直接抛出异常。调用方要确保异常一定会被 await、gather 或 TaskGroup 收集；不要用无人持有引用的后台集合逃避错误处理，否则还是会遇到“Task exception was never retrieved”的未捕获异常日志。

不要把所有 I/O 都改成 eager

网络请求、文件操作、锁竞争和限流等待本来就需要回到事件循环处理。全局启用后，收益很小的任务也会承担额外的语义变化成本。更稳妥的做法是先按同步完成比例筛选热点路径，再在隔离的事件循环或者明确的服务边界内试用。

把回归检查写成可重复的门禁

至少准备四组测试用例：缓存命中、真实等待、同步异常和取消。对命中场景看任务是否提前完成；对等待场景看超时和取消是否还能按预期传播；对异常场景看调用方在创建和等待两个边界都能正常收集到错误。

async def test_task_group_with_explicit_mode():
    async with asyncio.TaskGroup() as group:
        group.create_task(read_cache("hit"), eager_start=True)
        group.create_task(read_cache("miss"), eager_start=True)

# 验收项：
# 1. Python 版本满足项目运行矩阵
# 2. 命中与未命中分开统计
# 3. 结果顺序由业务字段决定
# 4. 异常、取消都能被调用方观察到
# 5. 恢复普通工厂后旧测试仍通过

如果项目需要自定义 Task 子类，标准库还提供 asyncio.create_eager_task_factory()，但自定义构造器必须符合 Task.__init__ 的签名并返回兼容对象。除非确实要接入任务追踪逻辑，否则优先用内置工厂更容易定位问题。

一份可以落地的迁移清单

• 锁定解释器版本，并在 CI 中显式运行 eager 与普通工厂两套关键测试。
• 先找同步完成比例高、结果不依赖创建顺序的协程；不要从所有后台任务开始批量修改。
• 给结果增加业务序号或者业务键，禁止用任务完成先后代替业务排序。
• 让 TaskGroup、gather 或显式 await 持有任务并回收异常。
• 记录命中耗时、等待耗时、异常数和取消数，优化后没有实际收益就回滚工厂设置。

相关问题

eager_task_factory 是 Python 3.14 才有吗？

不是。官方文档标注它在 Python 3.12 加入；Python 3.14 主要扩展了任务创建参数的透传能力。部署环境仍应以项目实际支持的 Python 版本为准。

开启后网络请求会更快吗？

它减少的是协程启动和事件循环调度的开销，不会缩短网络服务端的响应时间。只有大量同步完成或缓存命中的短协程，才值得单独测量收益。

可以只让一个任务 eager 吗？

可以在支持该参数的任务创建路径中显式传入 eager_start，也可以把工厂配置限制在单独的事件循环里。使用前先确认目标 Python 版本和任务组 API 的签名。

为什么测试日志顺序变了？

因为 eager 模式会在任务构造阶段立即运行协程，完成或异常都可能早于后续任务创建。日志断言应关注事件字段和因果关系，不要只比较输出行的排列顺序。

对 Python 异步服务来说，eager 工厂更像一次调度语义迁移：缓存命中可以少走一段调度路径，但任务顺序、异常时机和取消观察都需要重新验收。先用小范围基准测试验证真实收益，再用四组回归用例覆盖边界场景，通常比全局切换更稳妥。