FastAPI BackgroundTasks 阻塞的问题处理

FastAPI 中的 BackgroundTasks 允许在后台执行长时间运行的任务，不阻塞主线程，这对于不想阻塞用户请求非常有用。但是有时候我们的后台任务会出现阻塞，导致请求也被阻塞。本文将介绍这方面的问题和解决方法。

什么是 BackgroundTasks

BackgroundTasks是 FastAPI 框架的一个功能，它允许我们在响应已经返回给客户端后继续处理任务，而无需等待任务完成。这意味着这些任务将在后台异步执行，而不会阻塞主请求——响应循环。这在处理一些非实时关键的任务时非常有用，比如发送电子邮件、日志记录、消息通知等。

常见场景和问题

在以下情况下，BackgroundTasks特别有用：

发送电子邮件或短信通知
将数据写入异步日志
处理长时间运行的任务，如图像处理、文件转换等
异步地执行数据库操作

然而，使用BackgroundTasks时，我们需要注意以下问题：

异步任务是否会对应用程序的性能产生负面影响？
任务执行失败后，我们如何处理错误？
在某些场景下，任务的执行顺序是否很重要？

BackgroundTasks 使用

FastAPI 中通过 BackgroundTasks 类创建后台任务。简单使用如下:

from fastapi import BackgroundTasks

def write_notification(email, message):
    pass

@app.post("/send-notification")
async def send_notification(email: str, background_tasks: BackgroundTasks): 
    background_tasks.add_task(write_notification, email, message)
    return {"message": "Notification sent in the background"}

这样 write_notification 函数会在后台执行，不阻塞主线程。更多详细用法参考 FastAPI 官方文档。

阻塞问题

但是有时候我们的后台函数中也可能包含阻塞操作，例如网络请求:

# 后台任务函数
def write_notification(email, message):
    resp = requests.post("http://notification-service/send", json={"email": email, "message": message}) # 网络请求会阻塞
    print(resp.text)

这是因为后台函数里的网络请求阻塞了，会导致整个请求也被阻塞。

解决方案

要解决这个问题，可以通过线程、进程或协程来执行后台任务:

线程 - 使用 threading 模块:

import threading

def write_notification(email, message):
    t = threading.Thread(target=send_notification, args=[email, message])  
    t.start()

2. 进程 - 使用 multiprocessing 模块:

import multiprocessing

def write_notification(email, message):
    p = multiprocessing.Process(target=send_notification, args=[email, message])
    p.start()

3. 协程 - 使用 asyncio 模块:

import asyncio

async def write_notification(email, message):
    await notify(email, message) # 使用 asyncio.sleep 代替 time.sleep

def background_tasks(email, message):
    loop = asyncio.get_event_loop()
    loop.create_task(write_notification(email, message))

提示和注意事项

尽量将任务保持简短，避免执行长时间运行的阻塞任务，以免影响应用程序性能。
注意错误处理。如果异步任务失败，确保有合适的错误处理机制来记录和处理错误。

总结

FastAPI 的后台任务也可能发生阻塞，解决阻塞的方法可以通过线程、进程或协程来避免，推荐的方案是使用 asyncio + async/await，避免后台任务中含有阻塞操作如网络请求

使用 Apifox 调试 FastAPI 接口

如果你是 FastAPI 开发者，你经常需要与 API 打交道，确保你的应用程序能够正常工作。这时，一个强大的接口测试工具就会派上用场。

Apifox 是一个集 API 文档、API 调试、API Mock 和 API 自动化测试于一体的 API 协作平台，我们可以通过 Apifox 来更方便的调试 FastAPI。

如果想快速的调试一条接口，新建一个项目后，在项目中选择“调试模式”，填写请求地址后即可快速发送请求，并获得响应结果，如图所示：

立即体验 Apifox

知识扩展

了解更多关于 FastAPI 的知识：

参考