本文介绍如何在 FastAPI 应用中真正实现「发后即忘」式长时任务执行——通过 ThreadPoolExecutor + loop.run_in_executor 调度 asyncio 任务，避免阻塞主事件循环，同时兼顾 ffmpeg 等外部命令调用、文件处理与数据库写入的完整性与可观测性。
本文介绍如何在 fastapi 应用中真正实现「发后即忘」式长时任务执行——通过 `threadpoolexecutor` + `loop.run_in_executor` 调度 `asyncio` 任务，避免阻塞主事件循环，同时兼顾 ffmpeg 等外部命令调用、文件处理与数据库写入的完整性与可观测性。
在构建高并发 API 服务时，一个常见但棘手的问题是：如何安全执行耗时操作（如视频转码、大文件处理、模型推理等），而不拖垮 FastAPI 的 asyncio 主事件循环？直接 await 长时协程（尤其是内部含同步 I/O 或 CPU 密集型逻辑的协程）会导致整个应用响应延迟甚至超时。虽然 asyncio.subprocess 可以异步启动 ffmpeg，但 process.communicate() 仍是协程挂起点——若任务持续数分钟，它将长期占用事件循环资源。
正确的解法不是“更深度异步化”，而是明确任务边界并合理委托执行环境：将真正耗时的部分（含 await 的完整协程链）交由线程池托管，让主线程/事件循环立即返回，后续通过回调、状态轮询或消息通知机制处理结果。
✅ 推荐架构：生命周期管理的线程池 + 协程封装调度
以下是一个生产就绪的三段式实现方案，已验证兼容 gunicorn + uvicorn 部署模式：
1. 全局线程池注入生命周期（lifespan）
使用 @asynccontextmanager 在应用启动时初始化线程池，并确保优雅关闭：
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from fastapi import FastAPI
POOL_MAX_THREADS = 20
@asynccontextmanager
async def lifespan(app: FastAPI):
# 初始化共享线程池（非全局变量，避免多 worker 冲突）
app.state.pool = ThreadPoolExecutor(max_workers=POOL_MAX_THREADS)
# 启动时预加载或健康检查（可选）
await on_startup_single(app)
yield # 请求期间 pool 可被访问
# 关闭前等待任务完成（可选 timeout）
app.state.pool.shutdown(wait=True, cancel_futures=False)创建 FastAPI 实例时传入该 lifespan：
app = FastAPI(lifespan=lifespan)
⚠️ 注意：切勿使用 concurrent.futures.ProcessPoolExecutor 直接运行含 asyncio 的协程（如 asyncio.run(...)），因其子进程无事件循环上下文，会抛出 RuntimeError: There is no current event loop in thread。线程池是唯一安全载体。
2. 端点层：透传线程池至业务逻辑
通过 Request.state 将线程池实例传递给 CRUD 层，保持依赖显式、测试友好：
from fastapi import APIRouter, Request, Body
from pydantic import BaseModel
router = APIRouter()
class EncodeRequest(BaseModel):
input_path: str
output_path: str
preset: str = "fast"
@router.post("/encode")
async def start_encoding(request: Request, payload: EncodeRequest):
# 异步触发任务，立即返回 202 Accepted
task_id = await crud.start_video_encoding(payload, request.app.state.pool)
return {"task_id": task_id, "status": "accepted"}3. 业务层：封装协程并提交至线程池
关键在于：用 loop.run_in_executor 执行一个包装函数，该函数内调用 asyncio.run() 运行完整异步任务链。这是绕过“线程内无事件循环”限制的标准模式：
import asyncio
import logging
from uuid import uuid4
from typing import Dict, Any
# 全局任务状态存储（生产中建议替换为 Redis / DB）
_TASK_REGISTRY: Dict[str, Dict[str, Any]] = {}
async def _run_ffmpeg_task(input_path: str, output_path: str, preset: str) -> dict:
"""完整异步任务：调用 ffmpeg → 处理输出 → 写库"""
try:
# 使用你已有的 ffmpeg-python 异步封装（注意：run_async_async 必须支持 run=True）
from your_ffmpeg_module import run_async_async
result = await run_async_async(
ffmpeg.input(input_path).output(output_path, preset=preset),
run=True
)
if result.returncode != 0:
raise RuntimeError(f"FFmpeg failed: {result.stderr[:200]}")
# 后续同步/异步操作（如文件校验、DB 写入）
await save_encoding_result_to_db(output_path, result)
return {"success": True, "output": output_path}
except Exception as e:
logging.error(f"Task failed: {e}")
return {"success": False, "error": str(e)}
def _sync_wrapper(input_path: str, output_path: str, preset: str) -> dict:
"""线程内同步入口：必须在此启动新事件循环"""
return asyncio.run(_run_ffmpeg_task(input_path, output_path, preset))
async def start_video_encoding(payload: EncodeRequest, pool: ThreadPoolExecutor) -> str:
"""对外暴露的异步接口：提交任务并注册状态"""
task_id = str(uuid4())
loop = asyncio.get_running_loop()
# ✅ 核心：run_in_executor + asyncio.run 组合
future = loop.run_in_executor(
pool,
_sync_wrapper,
payload.input_path,
payload.output_path,
payload.preset
)
# 可选：记录 future 用于后续取消/监控（需额外设计）
_TASK_REGISTRY[task_id] = {
"future": future,
"started_at": asyncio.get_event_loop().time(),
"status": "running"
}
# fire-and-forget：不 await，立即返回
asyncio.create_task(_handle_task_completion(task_id, future))
return task_id
async def _handle_task_completion(task_id: str, future: asyncio.Future):
"""后台任务：等待完成、更新状态、触发通知（如 WebSocket / webhook）"""
try:
result = await future
_TASK_REGISTRY[task_id].update({"status": "completed", "result": result})
await notify_completion(task_id, result) # e.g., via Redis Pub/Sub
except Exception as e:
_TASK_REGISTRY[task_id].update({"status": "failed", "error": str(e)})
logging.error(f"Task {task_id} crashed: {e}")? 关键注意事项与最佳实践
- 不要在 run_in_executor 中直接 await：run_in_executor 只接受同步函数。若需在子线程中跑协程，必须用 asyncio.run() 包裹。
- 避免共享状态竞争：_TASK_REGISTRY 仅为演示，生产环境务必使用线程安全结构（如 threading.Lock）或外部存储（Redis 原子操作）。
- 资源隔离优于复用：每个长时任务应有独立临时目录、超时控制与错误重试策略，防止 ffmpeg 挂起导致线程池耗尽。
- 监控与可观测性：记录任务 ID、启动时间、执行时长、退出码；集成 Prometheus 指标（如 encoding_tasks_running_total）。
-
替代方案对比：
- celery：适合复杂工作流与任务持久化，但引入额外运维成本；
- asyncio.Queue + 后台任务：适用于轻量级队列消费，但无法解决单个任务阻塞问题；
- multiprocessing：适用于纯 CPU 密集型任务，但无法直接运行 asyncio 代码，且进程间通信开销大。
✅ 总结
FastAPI 本身不提供内置的后台任务调度器，但通过 lifespan 管理线程池 + run_in_executor 封装 asyncio.run，即可构建轻量、可控、符合 ASGI 规范的异步长时任务系统。该方案既规避了事件循环阻塞风险，又保留了协程的开发便利性与生态兼容性（如 ffmpeg-python、httpx、aiosqlite 等），是中小型视频/媒体服务的理想选择。
