一、FastAPI异步与后台处理的核心价值

在传统同步Web应用架构中，当用户发起请求时，服务器需完全处理完当前请求才能响应下一个请求。面对发送邮件、视频转码、批量数据处理等耗时操作时，这种模式会导致请求队列阻塞，服务器响应延迟，用户体验大幅下降。

FastAPI凭借异步编程模型与后台任务机制，完美解决了这一痛点：

• 异步编程：基于Python的async/await语法，服务器在等待I/O操作（如数据库查询、API调用）时，可释放控制权处理其他请求，最大化利用系统资源。

• 后台任务：将耗时操作剥离至后台执行，前端可立即收到响应，无需等待任务完成，显著提升API的响应速度与吞吐量。

• 并发处理：通过协程调度实现多任务交替执行，避免线程切换的性能开销，在高并发场景下优势尤为明显。

二、FastAPI异步编程基础

2.1 异步路径操作函数

使用async def声明异步路径操作函数，函数内部可通过await调用异步操作，让服务器在等待过程中处理其他请求。示例如下：

from fastapi import FastAPI
import asyncio

app = FastAPI()

# 模拟异步数据库查询
async def fetch_item_from_db(item_id: int):
   await asyncio.sleep(1)  # 模拟I/O等待
   return {"item_id": item_id, "name": "示例商品", "price": 99.9}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
   item = await fetch_item_from_db(item_id)
   return item


当客户端访问/items/1时，服务器在执行await fetch_item_from_db(item_id)时会释放事件循环，可同时处理其他请求，待异步操作完成后再返回结果。

2.2 同步函数的异步兼容

对于不支持异步的第三方库或同步代码，FastAPI会自动将其放入单独的线程池中运行，避免阻塞主事件循环。示例：

import time
from fastapi import FastAPI

app = FastAPI()

# 同步耗时操作
def heavy_sync_operation():
   time.sleep(3)  # 模拟CPU密集型或同步I/O操作
   return "同步操作完成"

@app.get("/sync-operation")
def sync_operation():
   result = heavy_sync_operation()
   return {"message": result}


此处使用普通def定义路径操作函数，FastAPI会自动在线程池中执行heavy_sync_operation，主事件循环不会被阻塞。

三、后台任务实战指南

3.1 基本后台任务实现

FastAPI提供BackgroundTasks类，可轻松将任务放入后台执行。以下是发送通知的示例：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

# 后台任务函数：写入通知日志
def write_notification(email: str, message: str):
   with open("notification_log.txt", "a", encoding="utf-8") as f:
       log_content = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 通知至{email}: {message}\n"
       f.write(log_content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
   # 添加后台任务
   background_tasks.add_task(write_notification, email, message="欢迎使用FastAPI后台任务！")
   return {"message": "通知已在后台发送，请注意查收"}


当客户端调用/send-notification/user@example.com时，接口会立即返回响应，write_notification函数则在后台执行，完成日志写入操作。

3.2 依赖注入与后台任务结合

后台任务可与FastAPI的依赖注入系统结合，实现更灵活的任务调度。示例如下：

from fastapi import BackgroundTasks, Depends, FastAPI
import time

app = FastAPI()

# 后台任务：记录查询日志
def write_query_log(query: str):
   with open("query_log.txt", "a", encoding="utf-8") as f:
       log_content = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 查询参数: {query}\n"
       f.write(log_content)

# 依赖函数：处理查询参数并添加后台任务
def get_query(background_tasks: BackgroundTasks, q: str = None):
   if q:
       background_tasks.add_task(write_query_log, q)
   return q

@app.get("/search")
async def search(q: str = Depends(get_query)):
   return {"message": f"已根据参数{q}执行搜索"}


当客户端访问/search?q=FastAPI时，依赖函数get_query会自动将查询日志任务添加至后台，接口无需等待日志写入即可返回结果。

四、长时任务的异步处理方案

对于视频转码、模型推理等耗时超过数十秒的任务，直接使用BackgroundTasks可能存在风险（如服务器重启导致任务中断）。此时可采用线程池+协程封装的方案，实现“发后即忘”的安全执行。

4.1 全局线程池的生命周期管理

通过@asynccontextmanager在应用启动时初始化线程池，确保优雅关闭：

from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio

POOL_MAX_THREADS = 20

# 应用生命周期管理
@asynccontextmanager
async def lifespan(app: FastAPI):
   # 初始化线程池
   app.state.pool = ThreadPoolExecutor(max_workers=POOL_MAX_THREADS)
   yield
   # 关闭线程池，等待任务完成
   app.state.pool.shutdown(wait=True, cancel_futures=False)

app = FastAPI(lifespan=lifespan)


4.2 长时任务的调度与执行

将长时任务封装为协程，通过线程池调度执行，避免阻塞主事件循环。示例为视频转码任务：

from fastapi import APIRouter, Request
from pydantic import BaseModel
import subprocess
import asyncio

router = APIRouter()

# 请求模型
class EncodeRequest(BaseModel):
   input_path: str
   output_path: str
   preset: str = "fast"

# 视频转码任务（同步函数）
def video_encode_task(input_path: str, output_path: str, preset: str):
   cmd = [
       "ffmpeg",
       "-i", input_path,
       "-c:v", "libx264",
       "-preset", preset,
       "-c:a", "aac",
       "-y",  # 覆盖输出文件
       output_path
   ]
   try:
       subprocess.run(cmd, check=True, capture_output=True, text=True)
       return {"status": "success", "message": "转码完成"}
   except subprocess.CalledProcessError as e:
       return {"status": "failed", "message": f"转码失败: {e.stderr}"}

# 异步封装：在线程池中执行同步任务
async def async_video_encode(input_path: str, output_path: str, preset: str, pool):
   loop = asyncio.get_event_loop()
   result = await loop.run_in_executor(
       pool,
       video_encode_task,
       input_path,
       output_path,
       preset
   )
   return result

# 转码接口
@router.post("/encode")
async def start_encoding(request: Request, payload: EncodeRequest):
   # 在线程池中调度转码任务
   task = asyncio.create_task(
       async_video_encode(
           payload.input_path,
           payload.output_path,
           payload.preset,
           request.app.state.pool
       )
   )
   return {"message": "转码任务已启动", "task_id": id(task)}


该方案通过线程池隔离长时任务，主事件循环可立即返回响应，任务在后台独立执行，同时支持任务状态查询、结果回调等扩展功能。

五、性能优化与注意事项

1. 合理设置线程池大小：线程池过大可能导致上下文切换开销增加，过小则无法充分利用系统资源。建议根据CPU核心数与任务类型调整，通常设置为CPU核心数*2。

2. 避免在异步函数中执行同步阻塞操作：若异步函数中存在time.sleep()、同步数据库查询等阻塞操作，会导致事件循环被阻塞，应使用asyncio.sleep()或异步库替代。

3. 任务结果的持久化与查询：对于重要任务，建议将任务状态、结果存储至数据库或缓存中，提供查询接口让客户端获取任务进度。

4. 异常处理与日志记录：后台任务需完善异常处理逻辑，通过日志记录任务执行状态，便于问题排查。

5. 避免使用ProcessPoolExecutor执行协程：子进程无事件循环上下文，直接运行协程会抛出RuntimeError，线程池是执行协程的安全载体。