FastAPI后台重复执行函数：fastapi-utils与asyncio方案对比咨询

方案1：fastapi-utils的repeat_every装饰器 #

示例实现 #

from fastapi import FastAPI
from fastapi_utils.tasks import repeat_every

app = FastAPI()
_STATUS = "initial"

@app.on_event("startup")
@repeat_every(seconds=3)
async def update_status():
    global _STATUS
    # 模拟状态更新逻辑
    _STATUS = "updated"

@app.get("/status")
async def get_status():
    return {"status": _STATUS}

核心特点与疑问解答 #

• 可读性优势：装饰器直接声明执行间隔，代码逻辑直观，无需手动编写循环和睡眠逻辑。
• 维护风险：该库自2020年3月后停止更新，意味着：
  • 无法适配FastAPI新版本的特性（如生命周期事件优化、依赖注入升级）
  • 潜在的兼容性问题（如Python 3.10+的asyncio新特性）无法得到修复
  • 社区支持缺失，遇到问题难以找到解决方案

方案2：asyncio.create_task原生实现 #

示例实现 #

import asyncio
from fastapi import FastAPI

app = FastAPI()
_STATUS = "initial"
background_task = None

async def update_status_loop():
    while True:
        global _STATUS
        # 模拟状态更新逻辑
        _STATUS = "updated"
        await asyncio.sleep(3)

@app.on_event("startup")
async def app_startup():
    global background_task
    background_task = asyncio.create_task(update_status_loop())

@app.on_event("shutdown")
async def app_shutdown():
    background_task.cancel()
    try:
        await background_task
    except asyncio.CancelledError:
        pass

@app.get("/status")
async def get_status():
    return {"status": _STATUS}

细节疑问解答 #

1. await asyncio.sleep(3)是否必要？
   绝对必要。asyncio基于单线程事件循环运行，若没有await让出控制权，while True会完全阻塞事件循环，导致FastAPI无法处理任何HTTP请求。sleep(3)是让事件循环得以处理其他任务的关键。

2. 替换为耗时2秒的函数，是否要改为sleep(1)？
   取决于你要的执行逻辑：

   • 若希望任务结束后间隔3秒再执行下一次：保持sleep(3)即可，总周期为任务耗时+3秒
   • 若希望固定每3秒触发一次任务（忽略任务耗时）：需要计算时间差，比如记录上次任务开始时间，每次 sleep 时长为3 - (当前时间 - 上次开始时间)，避免任务耗时打乱间隔
   • 若希望总周期严格为3秒：则改为sleep(1)（3秒总周期 - 2秒任务耗时）

3. 两种方案的app_startup函数区别？

   • repeat_every的startup：装饰器自动封装了任务创建、循环调度和异常处理，你只需编写核心业务逻辑
   • asyncio.create_task的startup：需要手动创建任务并保存引用，还要在shutdown事件中手动取消任务、捕获CancelledError，逻辑更繁琐但完全可控

优缺点对比 #

选择建议 #

1. 短期原型/小项目：可以用repeat_every，快速实现需求，只要当前FastAPI和Python版本与库兼容（如Python 3.9及以下、FastAPI 0.6x版本），但要做好后续重构的准备。
2. 生产环境/长期维护项目：强烈推荐asyncio.create_task原生实现。虽然代码量稍多，但完全可控，不会因第三方库停止更新陷入被动，还能根据业务需求灵活扩展（如动态调整任务间隔、添加失败重试）。
3. 补充：你提到全局变量_STATUS仅为示例，生产环境建议用依赖注入的状态容器（如Depends配合自定义类）或Redis/数据库存储状态，避免多进程/多实例下的状态不一致问题。

内容的提问来源于stack exchange，提问作者aura