1. 后台任务和轮询这对组合解决的核心问题如果你用 FastAPI 写过真实项目后台任务和轮询迟早会一起找上你。我之前就遇过这么个需求前端上传一批产品图片后端要调用第三方图像处理服务逐张压缩、加水印、生成缩略图。最开始我图省事直接在请求里同步处理结果图片一多接口动不动就要跑几十秒前端 Axios 默认超时直接断开用户反复点击提交服务端内存里积压了一堆重复任务。后来我改成“先收任务、后台慢慢跑、前端轮询状态”这个问题才算彻底解决。FastAPI 后台任务解决的本质上是“HTTP 请求-响应模型”和“长耗时操作”之间的矛盾。HTTP 协议设计之初就是为了短请求客户端发一个请求服务端算完立刻返回。一旦服务端要花几十秒甚至几分钟才能给结果连接就可能超时负载均衡器和网关也会介入断开连接。这时候把任务丢到后台执行先立刻返回一个task_id让客户端隔几秒再问一次“任务怎么样了”是成本最低、最容易实现的方案。轮询在这里承担的是一个“状态同步通道”的角色。客户端拿到task_id后不断调用一个状态查询接口服务端返回这个任务当前是排队中、执行中、已完成还是失败。这个模式看起来朴素但工程上非常稳不需要维持长连接不需要考虑反向代理对 WebSocket 的支持服务端也不需要在连接断开时做特殊清理。可能有人会问为什么不直接用 WebSocket 或者 Server-Sent EventsSSE做实时推送我也试过结论是如果团队里前端控制力不强、网络环境复杂、服务端有多个 worker 进程轮询的鲁棒性反而更高。WebSocket 和 SSE 在反向代理层需要额外配置超时和连接数而轮询就是普通的 GET 请求任何网络环境都能过。尤其是做企业级内部系统用户可能抱着老旧的浏览器前端安全策略又严格轮询是最不挑环境的选择。我常见的使用场景包括批量数据导入后的解析和清洗、报表生成、邮件群发、视频转码、调用大模型生成内容。这些业务的共同点是请求本身很轻真正重的活都在后头。用 FastAPI 后台任务接收请求把重活放进任务队列再通过轮询接口把进度暴露给前端整个系统配合下来非常顺畅。2. 先分清 BackgroundTasks 和 asyncio.create_taskFastAPI 里做后台任务有两条常见路线直接用 FastAPI 自带的BackgroundTasks或者用 asyncio 的create_task。我见过不少新手把这两个混着用结果代码风格混乱任务状态也管理不清。这一节我把两者的边界讲清楚。2.1 FastAPI 自带的 BackgroundTasks 到底帮你做了什么BackgroundTasks是 Starlette 提供的能力。你在路由函数里声明一个background_tasks: BackgroundTasks参数把耗时函数加进去等当前请求返回响应之后ASGI 服务器会在后台执行这些任务。看代码会更直观from fastapi import FastAPI, BackgroundTasks app FastAPI() def process_report(report_id: int): # 模拟长耗时操作 import time time.sleep(30) print(freport {report_id} processed) app.post(/reports) async def create_report(report_id: int, background_tasks: BackgroundTasks): background_tasks.add_task(process_report, report_id) return {message: 任务已提交, task_id: report_id}这段代码有个重要的隐藏逻辑BackgroundTasks是在响应已经发送给客户端之后才执行的。如果客户端在请求进行中崩溃任务依然会继续执行因为它不依赖客户端连接。这一点和直接await完全不同。但BackgroundTasks在轮询场景里有一个致命短板它没有一个全局唯一的 task 句柄。你可以自己定义task_id但BackgroundTasks本身不帮你维护任务状态它只是“执行完就完事”。你想知道任务当前到哪一步、成功还是失败它不管。所以如果你的业务只是“请求结束后发一封邮件”这种不关心结果的后台动作用它很合适一旦要轮询你就得自己在外面套一层状态存储。2.2 asyncio.create_task 的自由度与代价另一条路是用asyncio.create_task创建真正的异步任务。代码长这样import asyncio import uuid from fastapi import FastAPI app FastAPI() tasks {} async def long_job(task_id: str): tasks[task_id] {status: running, progress: 0} try: for i in range(10): await asyncio.sleep(1) tasks[task_id][progress] (i 1) * 10 tasks[task_id][status] success except Exception as e: tasks[task_id][status] failed tasks[task_id][error] str(e) app.post(/jobs) async def create_job(): task_id str(uuid.uuid4()) tasks[task_id] {status: pending, progress: 0} asyncio.create_task(long_job(task_id)) return {task_id: task_id} app.get(/jobs/{task_id}) async def get_job(task_id: str): return tasks.get(task_id, {status: not found})asyncio.create_task返回一个Task对象你可以持有它、查询它、取消它。我实战中经常用它来配合轮询接口因为它能让我控制任务启动的时机、追踪任务状态、把结果写到共享存储里。代价则是你要自己处理异常、自己清理任务记录不然字典会越来越大。这里有个很关键的区别BackgroundTasks更适合于“用完即走的后台副作用”asyncio.create_task适合“需要全程跟踪的长任务”。轮询场景绝大多数属于后者所以我的轮询项目都是用asyncio.create_task或者外部队列来承担核心逻辑。2.3 结合轮询场景的选型表格我用实际经历总结了一张选型表供你直接参考维度BackgroundTasksasyncio.create_task执行时机响应发送完成后调用后立刻调度能否拿到任务句柄不能能状态跟踪能力弱需自己写强可更新共享状态异常处理需要内部自行捕获可统一捕获并记录适合场景邮件通知、日志上报、简单清理长任务执行 轮询状态多 worker 支持不跨进程不跨进程如果你用了--workers 4启动 Uvicorn这两者在多进程环境下都不共享内存状态。这是第三个需要解决的问题——任务存储我在下一章细讲。3. 任务状态存储决定轮询的可靠性从内存到 Redis轮询接口看起来很简单但实际可靠与否几乎完全取决于任务状态存在哪里。我最早把状态存在 Python 内存字典里单机单 worker 跑没问题一旦上多 worker 或者重启服务问题立刻暴露。3.1 第一步单机内存 TaskManager 长什么样先看一个最基础的内存版任务管理器。我受够了一堆散落的字典操作通常会把任务状态封装成一个类import uuid import asyncio from enum import Enum from datetime import datetime class TaskStatus(str, Enum): PENDING pending RUNNING running SUCCESS success FAILED failed class MemoryTaskManager: def __init__(self): self._tasks {} def create(self): task_id str(uuid.uuid4()) self._tasks[task_id] { status: TaskStatus.PENDING, progress: 0, result: None, error: None, created_at: datetime.now().isoformat(), updated_at: datetime.now().isoformat(), } return task_id def update(self, task_id, **kwargs): task self._tasks.get(task_id) if task: task.update(kwargs) task[updated_at] datetime.now().isoformat() def get(self, task_id): return self._tasks.get(task_id)用这个类配合asyncio.create_task就能搭出一个最简单可用的轮询服务。单机、单进程、调试阶段这套方案完全够用。我也确实是先用它验证了整体流程再决定要不要换存储。3.2 多 worker 和重启带来的状态丢失当你用 Uvicorn 启动多个 worker或者使用 Gunicorn 管理多个进程内存字典的方案就失效了。原因很简单不同 worker 是不同进程进程之间的 Python 字典不可见。请求落到 worker A 创建任务轮询时负载均衡把请求转到 worker BB 的内存里根本没有这个任务。你会看到接口时而返回成功时而返回not found。重启服务就更严重内存一清空所有进行中的任务全没了。用户拿着之前的task_id来轮询只会得到“任务不存在”这是非常糟糕的体验。我后来把任务状态迁到了 Redis用 Hash 结构存每个任务的状态。核心改动其实不大import redis r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def create_task(): task_id str(uuid.uuid4()) r.hset( ftask:{task_id}, mapping{ status: pending, progress: 0, created_at: datetime.now().isoformat(), } ) return task_id轮询接口读的时候直接r.hgetall(ftask:{task_id})写的时候用r.hset更新字段。Redis 天然跨进程、跨连接共享状态多个 worker 都能读到同一份数据这才算真正解决了轮询的一致性问题。不过 Redis 方案也要注意一个细节别忘了给任务记录设置过期时间比如r.expire(ftask:{task_id}, 3600)否则一天跑几千个任务Redis 内存会被历史任务撑爆。轮询场景的任务结果一旦被前端拿到通常就不需要长期保存了设置一个 TTL 是必要的。3.3 推荐的项目目录结构既然涉及任务管理、路由、状态存储、后台执行项目就不能全都堆在main.py里。我目前比较顺手的目录结构是这样的app/ main.py # FastAPI 实例路由注册启动事件 api/ routes/ tasks.py # 任务创建、轮询查询、取消任务 dependencies.py # 公共依赖比如 Redis 连接 services/ task_manager.py # 任务状态管理封装 Redis 操作 worker.py # 具体的后台任务处理函数 models/ task.py # 状态枚举、数据结构定义 schemas/ task.py # Pydantic 请求/响应模型把任务创建路由和任务处理逻辑拆开最大的好处是排查问题时能快速定位。比如发现任务执行有 bug就去看worker.py发现轮询结果和预期不符就去看task_manager.py和tasks.py。状态枚举丢在models/task.py里前端和后端可以约定一套固定的文案和编号减少沟通成本。4. 状态轮询接口协议设计字段、频率、版本号任务在后台跑起来了状态也存好了接下来要设计轮询接口协议。这个设计很影响前后端联调体验也直接影响数据库或 Redis 的压力。4.1 一次干净的状态查询长什么样我建议状态查询接口的返回结构保持稳定尽量不要随意变字段。下面是经验之谈class TaskStatusResponse(BaseModel): task_id: str status: str # pending / running / success / failed progress: int # 0-100 message: str | None # 可附加信息比如当前处理到第几张图 result: dict | None # 任务成功时返回结果失败时为空 error: str | None # 失败原因 updated_at: str这样返回的好处是前端拿到status就能走分支逻辑。progress用于渲染进度条result只在成功时有值error只在失败时有值。不要试图把所有结果都塞进message字段那样解析起来很难受。4.2 版本号轮询只拉差异别把接口当数据库我看到很多团队做轮询前端每 2 秒拉一次完整数据后端查询数据库这在小规模没问题但并发一高就很浪费。这里可以用“版本号轮询”的思路优化。版本号轮询的说法来自客户端更新检查思路是服务端每次更新任务状态时把version字段加 1。客户端保存上次拿到的version下次轮询时把version带上服务端判断版本号没有变化就返回一个轻量级响应比如{version: 8, changed: false}只有版本号变化了才返回完整状态。这样可以大幅降低无效数据传输。结合 FastAPI 的 Query 参数来看app.get(/jobs/{task_id}) async def get_job(task_id: str, version: int 0): task task_manager.get(task_id) if task is None: return {changed: False, error: task not found}, 404 if task[version] version: return {changed: False, version: task[version]} return { changed: True, task: task, version: task[version], }如果任务状态更新频繁还可以用 HTTP 条件请求返回ETag或者304 Not Modified。这个方案可以少传很多无意义 JSON尤其是任务结果比较大比如包含多条生成文本或者图片 URL 时收益非常明显。4.3 轮询频率的工程决策轮询间隔没有标准答案。我以前喜欢固定 2 秒一次后来发现分场景会更合理任务时长在 10 秒以内轮询间隔可以设为 1 秒体验接近实时。任务时长在 1 分钟以上轮询间隔设为 3 到 5 秒就够了。任务时长以小时计算前端根本不用轮询改成“通知用户回来查看”更合理。另外强烈建议做指数退避。比如前端第一次等 1 秒没结束就等 2 秒、4 秒、8 秒封顶 10 秒。这样既能在任务刚提交时及时看到状态变化又不会在长时间任务里疯狂请求服务端。配合版本号轮询指数退避之后单个用户对服务器的 QPS 会降得比较低。哪怕同时在线几百个用户状态查询接口的压力也可控。5. 这些坑轮到你头上之前先记住解决方案后台任务和轮询写起来不难难的是那些隐性问题。我说三个我自己切切实实踩过的坑全是在生产环境里被用户和监控逼出来的。5.1 uvicorn 日志丢失感觉任务没跑其实是日志没落盘很多人在后台任务里写print(task started)然后用 Uvicorn 启动服务。前台执行时日志能看到换成后台任务后发现日志没了第一反应就是“任务没跑”。实际上任务跑了只是print的输出没有到你预期的地方。Uvicorn 自己对日志有一套处理默认情况下它会接管 logging 配置而print在这种配置下可能不会实时刷新。加上后台任务是在请求处理之外执行日志缓冲可能一直没有刷出来。我建议后台任务内部不用print而是用标准 logging 模块import logging logger logging.getLogger(app.task) def long_task(task_id: str): logger.info(task %s started, task_id) try: # do something logger.info(task %s finished, task_id) except Exception: logger.exception(task %s failed, task_id)然后确保在main.py里把日志配置好。一个常见做法是import logging logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s, )如果用了多个 worker还要考虑日志文件被多个进程同时写入的问题。我后来统一把日志交给了 Logstash 或者直接打到 stdout 由容器平台收集尽量避免本地文件多进程写冲突。5.2 重复提交和并发执行前端轮询有个常见衍生问题用户等得不耐烦疯狂点“提交”按钮同一份任务被创建了好几遍。服务端如果不做幂等控制后台会同时跑好几个相同的长任务消耗数据库连接和第三方接口配额。我处理这个问题时用了两种方式第一业务层幂等。用户提交任务时带一个request_id这个值由前端生成后端用request_id做唯一键。如果同一个request_id已经存在就直接返回已有任务不重复创建。app.post(/jobs) async def create_job(request_id: str Header(...)): existing task_manager.find_by_request_id(request_id) if existing: return {task_id: existing[task_id], duplicated: True} task_id task_manager.create(request_id) # start task return {task_id: task_id, duplicated: False}第二任务级互斥。比如同一个用户同时只允许一个转码任务在跑创建新任务前先查一下是否有运行中的任务有则直接拒绝。幂等控制看着增加了一点代码量实际上救了我很多次。尤其是在手机网络不稳定的场景下客户端超时自动重试是很常见的没有幂等后端会收到一堆重复请求。5.3 任务重启丢失与恢复策略后台任务跑了 10 分钟服务重启任务直接没了。如果这是离线报表任务还好用户重新提交就行如果是支付回调、订单处理这类关键任务丢任务就是事故。我的建议是重要任务不要只存在内存里至少要往 Redis 或数据库里写一份“任务元数据 中间状态”并且把耗时的每一步设计成可恢复的。简单做法是给任务加超时和重试标记def task_manager.get_stale_tasks(max_age_seconds600): # 找出所有 running 状态且 updated_at 超过阈值的任务 ...服务启动时扫描这些“卡住”的任务把它们重置为 pending或者标记为 failed 并记录错误原因。恢复策略没有银弹但至少要让任务状态在服务重启后是确定的而不是无影无踪。这个设计麻烦一点但一旦你负责的是核心业务这个成本是必须付出的。6. 进阶玩法让后台跑 LLM Agent前端轮询拿结果现在不少应用接入了 LangChain、LangGraph 这类框架让大模型 Agent 完成复杂任务比如自动分析文件、调用工具、多轮推理。这类任务耗时更不稳定有时 10 秒有时几分钟后端如果用同步请求处理用户很容易超时崩溃。把后台任务和轮询机制用到 Agent 调度上恰好是非常合适的一种架构。6.1 LangChain/LangGraph 长 Agent 任务如何接入这套模式我做过一个基于 FastAPI 和 LangGraph 的知识库问答服务用户提交一个问题后Agent 需要去搜索内部资料、调用向量检索、整理答案有时候还要访问数据库。整个链路跑下来经常超过 30 秒我最后就是把它封装成后台任务的。核心思路是请求进来创建任务记录状态置为 pending。用asyncio.create_task启动 Agent 执行流程。在 Agent 每步执行后更新任务状态比如正在检索资料、正在分析结果。前端轮询状态接口拿到当前 Message 展示给用户。代码骨架大致是async def run_agent(task_id: str, question: str): task_manager.update(task_id, statusrunning, messageAgent 已启动) try: # 以可轮询的方式执行 agent async for chunk in graph.astream({question: question}): task_manager.update( task_id, messagechunk.get(node, running), progressapply_some_progress(), ) final_answer ... task_manager.update(task_id, statussuccess, resultfinal_answer) except Exception as e: task_manager.update(task_id, statusfailed, errorstr(e))这么做还有个额外好处如果 Agent 还需要用户确认或者追加输入轮询接口也能把这些“等待输入”的状态暴露出去而不需要实时通信链路。Agent 应用通常在乎的是最终结果不是中间每一帧动画所以轮询完全够用。6.2 和 Gradio 一起部署时的注意点有些团队会把 FastAPI 和 Gradio 部署在同一个服务里Gradio 负责交互界面FastAPI 负责接业务请求。这时候要注意两点。第一Gradio 和 FastAPI 可以挂载在同一个应用上Gradio 的app.mount会占用根路径下的路由需要在挂载时指定前缀比如/gradio。轮询接口不要和 Gradio 内部路径冲突。第二Gradio 自己的queue机制和 FastAPI 后台任务是两个体系。如果你只把 Gradio 当展示层真正任务放 FastAPI 后台执行那状态数据还是走你自己的轮询接口不要依赖 Gradio 内部队列的状态。做过一次混合部署后我的结论是把 FastAPI 当作调度核心Gradio 只负责前端展示和事件触发两者通过 Redis 或数据库共享任务状态。这样以后想换成 Vue 前端或小程序后端完全不用动。6.3 我最后想说的几句话后台任务和轮询这套模式说到底是“异步化”思想在 HTTP 服务里的落地。FastAPI 给了你便捷的BackgroundTasks也给了你底层的asyncio.create_task但真正让系统可靠运行的是你对任务状态、并发控制、存储和日志的理解。我在实际项目里吃过不少亏最深刻的体会是先想清楚状态存哪里、怎么恢复再写任务执行逻辑。脑子里把“接口接收请求”和“后台执行任务”拆成两个独立环节很多问题从一开始就能规避。轮询虽然看起来“土”但它足够简单简单到不容易在生产环境下出幺蛾子这就已经是很大的优势了。如果你现在正卡在一个长耗时接口上不妨先把任务丢到后台再加一个状态轮询接口你会立刻发现前端体验和系统稳定性都有明显改善。等业务规模再大一些再考虑引入真正的消息队列但原理依然是这一套先接收任务后台执行客户端轮询拿结果。