点击python编程从入门到实践，置顶 公众号重磅 python入门资料，第一时间送达
![]()
读完需要 6 分钟
速读仅需 2 分钟
/ 后台异步这一篇就够了 /
1
BackgroundTasks 使用场景
有时候我们需要在 request 执行之后继续一些操作，但终端并不需要等待这些操作完成才能收到 response 。我列举一些场景大家看一下：
1.在自动出票完成后需要向各 ota 平台自动发送行程单信息
2.在执行完购票后需要向各户发送邮件通知购票成功信息
3.收到客户端的文件之后对文件进行二次处理
4....
5....
这些操作都需要一定的处理时间，但与返回给终端的 response 并无直接关系这个时候就可以通过定义后台任务 BackgroundTasks 来实现这个功能。
2
BackgroundTasks 实战
2.1
添加参数
首先我们需要导入 BackgroundTasks，并在路径操作函数中添加 BackgroundTasks 类型的参数。
# -*- encoding: utf-8 -*-
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
@app.post("/send_info")
async def send_notification(email: str, background_tasks: BackgroundTasks):
"""
发送提醒任务
"""
pass
2.2
任务函数
任务函数是指：在需要创建一个在后台任务中实际执行的函数。任务函数是一个标准函数。这个函数可以是 async def 或者 普通 def 的函数。
这里创建一个把指定内容写入文件的函数。
# -*- encoding: utf-8 -*-
def write_notification(email: str, message=""):
"""
写入文件操作模拟任务
"""
with open("log_test.txt", mode="w") as email_file:
content = "notification for {}: {}".format(email,message)
email_file.write(content)
2.3
添加后台任务
最后需要把任务函数添加到后台任务中
# -*- encoding: utf-8 -*-
import time
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message=""):
"""
写入文件操作模拟任务
"""
time.sleep(5)
with open("log_test.txt", mode="w") as email_file:
content = "notification for {}: {}".format(email,message)
email_file.write(content)
print("write end")
time.sleep(2)
@app.post("/send_info")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "now: %s Notification sent in the background" % time.time()}
2.4
add_task 解释
.add_task()接收以下参数信息：1.在后台运行的任务函数(例如 write_notification)2.任意序列的参数信息(例如 email)3.任意键值对的参数信息(例如 message="some notification")4.我们故意在 write_notification 方法中加入等待时间 来验证对于客户端的返回
2.5
依赖注入
后台任务可以与依赖注入系统一起使用，可以在不同层级的依赖项中声明 BackgroundTasks 参数，若对依赖注入系统还不是很熟悉的小伙伴可以会看一下上篇内容，本节就不做过多的解释了
# -*- encoding: utf-8 -*-
from typing import Optional
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: Optional[str] = None):
if q:
message = f"found query: {q}\n" background_tasks.add_task(write_log, message)
return q
@app.post("/send_info")
async def send_notification(email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)):
message = f"message to {email}\n" background_tasks.add_task(write_log, message)
return {"message": "Message sent"}
1.若需要执行大量的后台计算而不必一定要在同一进程中运行，例如：它不需要共享内存，变量等，则可使用其他更大的工具，例如:celery、MQ 系列 都是可以选择的但这些往往需要更复杂的配置，例如:RabbitMQ、Redis 之类的消息作业队列管理器，但是它们允许在多个进程（尤其是多个服务器）中运行后台任务。
2.若需要从同一 FastAPI 应用访问变量和对象，或者需要执行一些小的后台任务 例如:发送电子邮件、短信消息等，则只需使用即可 BackgroundTasks。
我们启动项目看一下执行效果：
![]()
![]()
![]()
我坚信：思考问题的方法远大于具体解决问题的方案，让我们继续一路前行，下期再见！
原创不易，只愿能帮助那些需要这些内容的同行或刚入行的小伙伴，你的每次 点赞、分享 都是我继续创作下去的动力，我希望能在推广 python 技术的道路上尽我一份力量，欢迎在评论区向我提问，我都会一一解答，记得一键三连支持一下哦！
加入python学习交流微信群，请后台回复「入群」
往期推荐
python生产实战 python 闭包之庖丁解牛篇
大型fastapi项目实战 靠 python 中间件解决方案涨薪了
大型fastapi项目实战 高并发请求神器之aiohttp(下)
大型fastapi项目实战 高并发请求神器之aiohttp(上) [建议收藏]
本文分享自微信公众号 - python编程军火库（PythonCoder1024）。
如有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。
