FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API，与 Starlette 和 Pydantic 集成。FastAPI 的设计理念之一是简洁且易于扩展。在处理耗时操作时，使用后台任务（Background Tasks）可以显著提升应用效率。本文将详细介绍 FastAPI 的后台任务功能，包括其原理、配置和使用方法。
一、后台任务原理
后台任务允许你在 FastAPI 应用中异步执行耗时操作。这些操作通常包括数据同步、批量处理、发送邮件等。后台任务使用异步编程模型，这意味着它们不会阻塞主线程，从而提高应用的响应速度和吞吐量。
FastAPI 使用
uvicorn
作为默认的 ASGI 服务器，而
uvicorn
支持异步任务队列。当你在 FastAPI 应用中定义后台任务时，它将被添加到任务队列中，并由
uvicorn
负责异步执行。
二、配置后台任务
要在 FastAPI 应用中启用后台任务，你需要执行以下步骤：
安装必要的库
：
pip install fastapi uvicorn[standard]
创建后台任务类
：
from fastapi import FastAPI
from fastapi.background import BackgroundTasks
app = FastAPI()
@app.post(“/tasks/”)
async def create_task(background_tasks: BackgroundTasks):
background_tasks.add_task(long_running_task)
return {“message”: “Background task added”}
在上述代码中，我们创建了一个名为
long_running_task
的异步函数，它将被添加到后台任务队列中。
定义后台任务函数
：
async def long_running_task():
# 这里是耗时操作的代码
await asyncio.sleep(10) # 模拟耗时操作
print(“Background task completed”)
在这个例子中，我们使用
asyncio.sleep(10)
来模拟一个耗时操作。
运行 FastAPI 应用
：
uvicorn main:app –reload
在这个例子中，
main
是你的 Python 模块名，
app
是 FastAPI 应用的实例。
三、使用 Celery 作为任务队列
虽然 FastAPI 支持内置的后台任务，但如果你需要更强大的任务队列功能，可以考虑使用 Celery。Celery 是一个异步任务队列/作业队列基于分布式消息传递的开源项目。
安装 Celery
：
pip install celery[redis]
配置 Celery
：
from celery import Celery
celery_app = Celery(
“tasks”,
broker=”redis://localhost:6379/0″,
backend=”redis://localhost:6379/0″,
)
@celery_app.task
def long_running_task():
# 这里是耗时操作的代码
await asyncio.sleep(10) # 模拟耗时操作
print(“Background task completed”)
在 FastAPI 应用中使用 Celery
：
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
@app.post(“/tasks/”)
async def create_task(background_tasks: BackgroundTasks):
background_tasks.add_task(long_running_task.delay())
return {“message”: “Background task added”}
在这个例子中，我们使用
long_running_task.delay()
来调用 Celery 任务。
四、总结
后台任务在 FastAPI 应用中发挥着重要作用，可以帮助你轻松应对耗时操作，提升应用效率。通过使用内置的后台任务或 Celery，你可以实现异步处理，避免阻塞主线程，从而提高应用的响应速度和吞吐量。希望本文能帮助你更好地理解和应用 FastAPI 的后台任务功能。