如何理解python大型项目后台异步

如何理解python大型项目后台异步，很多新手对此不是很清楚，为了帮助大家解决这个难题，下面小编将为大家详细讲解，有这方面需求的人可以来学习下，希望你能有所收获。

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">

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。

看完上述内容是否对您有帮助呢？如果还想对相关知识有进一步的了解或阅读更多相关文章，请关注亿速云行业资讯频道，感谢您对亿速云的支持。