如何理解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。

看完上述內容是否對您有幫助呢？如果還想對相關知識有進一步的了解或閱讀更多相關文章，請關注億速云行業資訊頻道，感謝您對億速云的支持。