如何理解python大型項目后臺異步，很多新手對此不是很清楚，為了幫助大家解決這個難題，下面小編將為大家詳細講解，有這方面需求的人可以來學(xué)習(xí)下，希望你能有所收獲。

成都創(chuàng)新互聯(lián)公司是一家專業(yè)提供金壇企業(yè)網(wǎng)站建設(shè),專注與成都網(wǎng)站制作、成都做網(wǎng)站、成都h5網(wǎng)站建設(shè)、小程序制作等業(yè)務(wù)。10年已為金壇眾多企業(yè)、政府機構(gòu)等服務(wù)。創(chuàng)新互聯(lián)專業(yè)網(wǎng)站設(shè)計公司優(yōu)惠進行中。

1

BackgroundTasks 使用場景

有時候我們需要在 request 執(zhí)行之后繼續(xù)一些操作，但終端并不需要等待這些操作完成才能收到 response 。我列舉一些場景大家看一下：

1.在自動出票完成后需要向各 ota 平臺自動發(fā)送行程單信息
2.在執(zhí)行完購票后需要向各戶發(fā)送郵件通知購票成功信息
3.收到客戶端的文件之后對文件進行二次處理
4....
5....

這些操作都需要一定的處理時間，但與返回給終端的 response 并無直接關(guān)系這個時候就可以通過定義后臺任務(wù) BackgroundTasks 來實現(xiàn)這個功能。

2

BackgroundTasks 實戰(zhàn)

2.1

添加參數(shù)

首先我們需要導(dǎo)入 BackgroundTasks，并在路徑操作函數(shù)中添加 BackgroundTasks 類型的參數(shù)。

# -*- encoding: utf-8 -*-
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

@app.post("/send_info")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    """
    發(fā)送提醒任務(wù)
    """
　　pass

2.2

任務(wù)函數(shù)

任務(wù)函數(shù)是指：在需要創(chuàng)建一個在后臺任務(wù)中實際執(zhí)行的函數(shù)。任務(wù)函數(shù)是一個標(biāo)準(zhǔn)函數(shù)。這個函數(shù)可以是 async def 或者 普通 def 的函數(shù)。

這里創(chuàng)建一個把指定內(nèi)容寫入文件的函數(shù)。

# -*- encoding: utf-8 -*-
def write_notification(email: str, message=""):
    """
    寫入文件操作模擬任務(wù)
    """
    with open("log_test.txt", mode="w") as email_file:
        content = "notification for {}: {}".format(email,message)
        email_file.write(content)

2.3

添加后臺任務(wù)

最后需要把任務(wù)函數(shù)添加到后臺任務(wù)中

# -*- encoding: utf-8 -*-
import time
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message=""):
    """
    寫入文件操作模擬任務(wù)
    """
    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()接收以下參數(shù)信息：1.在后臺運行的任務(wù)函數(shù)(例如 write_notification)2.任意序列的參數(shù)信息(例如 email)3.任意鍵值對的參數(shù)信息(例如 message="some notification")4.我們故意在 write_notification 方法中加入等待時間 來驗證對于客戶端的返回

2.5

依賴注入

后臺任務(wù)可以與依賴注入系統(tǒng)一起使用，可以在不同層級的依賴項中聲明 BackgroundTasks 參數(shù)

# -*- 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.若需要執(zhí)行大量的后臺計算而不必一定要在同一進程中運行，例如：它不需要共享內(nèi)存，變量等，則可使用其他更大的工具，例如:celery、MQ 系列 都是可以選擇的但這些往往需要更復(fù)雜的配置，例如:RabbitMQ、redis 之類的消息作業(yè)隊列管理器，但是它們允許在多個進程（尤其是多個服務(wù)器）中運行后臺任務(wù)。
2.若需要從同一 FastAPI 應(yīng)用訪問變量和對象，或者需要執(zhí)行一些小的后臺任務(wù) 例如:發(fā)送電子郵件、短信消息等，則只需使用即可 BackgroundTasks。

看完上述內(nèi)容是否對您有幫助呢？如果還想對相關(guān)知識有進一步的了解或閱讀更多相關(guān)文章，請關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝您對創(chuàng)新互聯(lián)的支持。