FastAPI后台任务与轮询机制实战指南

1. 后台任务和轮询这对组合解决的核心问题

如果你用 FastAPI 写过真实项目，后台任务和轮询迟早会一起找上你。我之前就遇过这么个需求：前端上传一批产品图片，后端要调用第三方图像处理服务逐张压缩、加水印、生成缩略图。最开始我图省事，直接在请求里同步处理，结果图片一多，接口动不动就要跑几十秒，前端 Axios 默认超时直接断开，用户反复点击提交，服务端内存里积压了一堆重复任务。后来我改成“先收任务、后台慢慢跑、前端轮询状态”，这个问题才算彻底解决。

FastAPI 后台任务解决的，本质上是“HTTP 请求-响应模型”和“长耗时操作”之间的矛盾。HTTP 协议设计之初就是为了短请求：客户端发一个请求，服务端算完立刻返回。一旦服务端要花几十秒甚至几分钟才能给结果，连接就可能超时，负载均衡器和网关也会介入断开连接。这时候把任务丢到后台执行，先立刻返回一个task_id，让客户端隔几秒再问一次“任务怎么样了”，是成本最低、最容易实现的方案。

轮询在这里承担的是一个“状态同步通道”的角色。客户端拿到task_id后，不断调用一个状态查询接口，服务端返回这个任务当前是排队中、执行中、已完成还是失败。这个模式看起来朴素，但工程上非常稳：不需要维持长连接，不需要考虑反向代理对 WebSocket 的支持，服务端也不需要在连接断开时做特殊清理。

可能有人会问，为什么不直接用 WebSocket 或者 Server-Sent Events（SSE）做实时推送？我也试过，结论是：如果团队里前端控制力不强、网络环境复杂、服务端有多个 worker 进程，轮询的鲁棒性反而更高。WebSocket 和 SSE 在反向代理层需要额外配置超时和连接数；而轮询就是普通的 GET 请求，任何网络环境都能过。尤其是做企业级内部系统，用户可能抱着老旧的浏览器，前端安全策略又严格，轮询是最不挑环境的选择。

我常见的使用场景包括：批量数据导入后的解析和清洗、报表生成、邮件群发、视频转码、调用大模型生成内容。这些业务的共同点是：请求本身很轻，真正重的活都在后头。用 FastAPI 后台任务接收请求，把重活放进任务队列，再通过轮询接口把进度暴露给前端，整个系统配合下来非常顺畅。

2. 先分清 BackgroundTasks 和 asyncio.create_task

FastAPI 里做后台任务有两条常见路线：直接用 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(f"report {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 结合轮询场景的选型表格

我用实际经历总结了一张选型表，供你直接参考：

如果你用了--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 B，B 的内存里根本没有这个任务。你会看到接口时而返回成功，时而返回not found。

重启服务就更严重，内存一清空，所有进行中的任务全没了。用户拿着之前的task_id来轮询，只会得到“任务不存在”，这是非常糟糕的体验。

我后来把任务状态迁到了 Redis，用 Hash 结构存每个任务的状态。核心改动其实不大：

import redis r = redis.Redis(host="localhost", port=6379, decode_responses=True) def create_task(): task_id = str(uuid.uuid4()) r.hset( f"task:{task_id}", mapping={ "status": "pending", "progress": "0", "created_at": datetime.now().isoformat(), } ) return task_id

轮询接口读的时候直接r.hgetall(f"task:{task_id}")，写的时候用r.hset更新字段。Redis 天然跨进程、跨连接共享状态，多个 worker 都能读到同一份数据，这才算真正解决了轮询的一致性问题。

不过 Redis 方案也要注意一个细节：别忘了给任务记录设置过期时间，比如r.expire(f"task:{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( level=logging.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_seconds=600): # 找出所有 running 状态且 updated_at 超过阈值的任务 ...

服务启动时，扫描这些“卡住”的任务，把它们重置为 pending，或者标记为 failed 并记录错误原因。恢复策略没有银弹，但至少要让任务状态在服务重启后是确定的，而不是无影无踪。这个设计麻烦一点，但一旦你负责的是核心业务，这个成本是必须付出的。

6. 进阶玩法：让后台跑 LLM Agent，前端轮询拿结果

现在不少应用接入了 LangChain、LangGraph 这类框架，让大模型 Agent 完成复杂任务，比如自动分析文件、调用工具、多轮推理。这类任务耗时更不稳定，有时 10 秒，有时几分钟，后端如果用同步请求处理，用户很容易超时崩溃。把后台任务和轮询机制用到 Agent 调度上，恰好是非常合适的一种架构。

6.1 LangChain/LangGraph 长 Agent 任务如何接入这套模式

我做过一个基于 FastAPI 和 LangGraph 的知识库问答服务，用户提交一个问题后，Agent 需要去搜索内部资料、调用向量检索、整理答案，有时候还要访问数据库。整个链路跑下来经常超过 30 秒，我最后就是把它封装成后台任务的。

核心思路是：

1. 请求进来，创建任务记录，状态置为 pending。
2. 用asyncio.create_task启动 Agent 执行流程。
3. 在 Agent 每步执行后更新任务状态，比如正在检索资料、正在分析结果。
4. 前端轮询状态接口，拿到当前 Message 展示给用户。

代码骨架大致是：

async def run_agent(task_id: str, question: str): task_manager.update(task_id, status="running", message="Agent 已启动") try: # 以可轮询的方式执行 agent async for chunk in graph.astream({"question": question}): task_manager.update( task_id, message=chunk.get("node", "running"), progress=apply_some_progress(), ) final_answer = ... task_manager.update(task_id, status="success", result=final_answer) except Exception as e: task_manager.update(task_id, status="failed", error=str(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，但真正让系统可靠运行的，是你对任务状态、并发控制、存储和日志的理解。

我在实际项目里吃过不少亏，最深刻的体会是：先想清楚状态存哪里、怎么恢复，再写任务执行逻辑。脑子里把“接口接收请求”和“后台执行任务”拆成两个独立环节，很多问题从一开始就能规避。轮询虽然看起来“土”，但它足够简单，简单到不容易在生产环境下出幺蛾子，这就已经是很大的优势了。

如果你现在正卡在一个长耗时接口上，不妨先把任务丢到后台，再加一个状态轮询接口，你会立刻发现前端体验和系统稳定性都有明显改善。等业务规模再大一些，再考虑引入真正的消息队列，但原理依然是这一套：先接收任务，后台执行，客户端轮询拿结果。