快速且高效地構(gòu)建API是現(xiàn)代軟件開(kāi)發(fā)的一個(gè)基礎(chǔ)方面。然而，由于數(shù)據(jù)處理、第三方服務(wù)調(diào)用或復(fù)雜計(jì)算等原因，經(jīng)常會(huì)遇到執(zhí)行時(shí)間較長(zhǎng)的API端點(diǎn)。面對(duì)這樣的情況，確保這些長(zhǎng)時(shí)間運(yùn)行的任務(wù)不會(huì)降低用戶體驗(yàn)或系統(tǒng)性能至關(guān)重要。本博客文章旨在指導(dǎo)您如何在FastAPI中管理長(zhǎng)時(shí)間運(yùn)行的任務(wù)。
長(zhǎng)時(shí)間運(yùn)行API端點(diǎn)的挑戰(zhàn)
- 用戶體驗(yàn)：響應(yīng)時(shí)間過(guò)長(zhǎng)可能會(huì)導(dǎo)致用戶體驗(yàn)差。
- 資源利用：長(zhǎng)時(shí)間運(yùn)行的任務(wù)可能會(huì)消耗大量系統(tǒng)資源，從而可能影響
- 其他任務(wù)的性能。
- 錯(cuò)誤處理：執(zhí)行時(shí)間長(zhǎng)的任務(wù)更容易出錯(cuò)，需要健壯的錯(cuò)誤處理機(jī)制。
管理長(zhǎng)時(shí)間運(yùn)行任務(wù)的最佳實(shí)踐
異步端點(diǎn)
您可以使用Python的async def語(yǔ)法在FastAPI中定義異步端點(diǎn)，這有助于I/O綁定操作。
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
# 執(zhí)行一個(gè)長(zhǎng)時(shí)間運(yùn)行的任務(wù)
return {"message": "done"}
后臺(tái)任務(wù)
FastAPI允許您運(yùn)行可以在響應(yīng)已發(fā)送后繼續(xù)處理的后臺(tái)任務(wù)。
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def long_running_task():
# 執(zhí)行一個(gè)長(zhǎng)時(shí)間運(yùn)行的任務(wù)
pass
@app.get("/")
async def read_root(background_tasks: BackgroundTasks):
background_tasks.add_task(long_running_task)
return {"message": "任務(wù)正在后臺(tái)運(yùn)行"}
使用Celery進(jìn)行分布式任務(wù)隊(duì)列
對(duì)于特別長(zhǎng)時(shí)間運(yùn)行的任務(wù)，您可以將它們卸載到像Celery這樣的任務(wù)隊(duì)列。
from fastapi import FastAPI
from celery import Celery
app = FastAPI()
celery_app = Celery('tasks', broker='pyamqp://guest@localhost//')
@celery_app.task
def long_running_task():
# 執(zhí)行一個(gè)長(zhǎng)時(shí)間運(yùn)行的任務(wù)
pass
@app.get("/")
def read_root():
long_running_task.apply_async()
return {"message": "任務(wù)正在后臺(tái)運(yùn)行"}
通知用戶的策略
一旦任務(wù)被卸載或變?yōu)楫惒?，通知用戶其完成狀態(tài)是必要的。以下是實(shí)現(xiàn)這一目標(biāo)的一些策略：
- 輪詢
在這種方法中，客戶端最初接收一個(gè)任務(wù)ID，然后反復(fù)輪詢一個(gè)端點(diǎn)以檢查任務(wù)的狀態(tài)。
服務(wù)端：
from fastapi import FastAPI
from some_task_queue import some_task_queue
app = FastAPI()
@app.post("/start_task/")
def start_task():
task_id = some_task_queue.enqueue("long_running_task")
return {"task_id": task_id}
@app.get("/get_result/{task_id}")
def get_result(task_id: str):
result = some_task_queue.get_result(task_id)
return {"result": result}
客戶端：
async function startAndPollTask() {
const response = await fetch('/start_task/');
const task = await response.json();
let result;
do {
const resultResponse = await fetch(`/get_result/${task.task_id}`);
result = await resultResponse.json();
if (result.is_done) {
break;
}
await new Promise(resolve => setTimeout(resolve, 2000));
} while(true);
console.log("Final result:", result);
}
- Webhooks
這里，客戶端提供一個(gè)回調(diào)URL，服務(wù)器可以在任務(wù)完成后將結(jié)果POST到這個(gè)URL。
服務(wù)端：
from fastapi import FastAPI, BackgroundTasks
import requests
app = FastAPI()
def long_running_task(callback_url):
# 執(zhí)行長(zhǎng)時(shí)間的任務(wù)
result = "some_result"
requests.post(callback_url, json={"result": result})
@app.post("/start_task/")
async def start_task(background_tasks: BackgroundTasks, callback_url: str):
background_tasks.add_task(long_running_task, callback_url)
return {"status": "任務(wù)已開(kāi)始"}
```
3. WebSockets
您可以在客戶端和服務(wù)器之間建立WebSocket連接，以在任務(wù)完成時(shí)發(fā)送結(jié)果。
服務(wù)端：
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket(“/ws/”)
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
# 長(zhǎng)時(shí)間運(yùn)行的任務(wù)
result = “some_result”
await websocket.send_json({“result”: result})
客戶端：
const socket = new WebSocket(‘ws://localhost:8000/ws/’);
socket.addEventListener(‘message’, function(event) {
const result = JSON.parse(event.data);
console.log(“Received result:”, result);
});
4. 服務(wù)器推送事件（SSE）
SSE允許服務(wù)器通過(guò)單個(gè)HTTP連接發(fā)送更新和最終結(jié)果。
服務(wù)端：
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.get(“/task_status/”)
def get_status():
def event_stream():
# 長(zhǎng)時(shí)間運(yùn)行的任務(wù)
result = “some_result”
yield f"data: {result}\n\n"
return StreamingResponse(event_stream(), media_type=“text/event-stream”)
```
客戶端：
const eventSource = new EventSource('/task_status/');
eventSource.onmessage = function(event) {
const result = event.data;
console.log("Received result:", result);
};
結(jié)論
長(zhǎng)時(shí)間運(yùn)行的任務(wù)在API設(shè)計(jì)中可能會(huì)帶來(lái)挑戰(zhàn)，但FastAPI提供了各種功能和技巧來(lái)有效地處理它們。無(wú)論是異步編程、后臺(tái)任務(wù)，還是像Celery、Webhooks和WebSockets這樣的高級(jí)策略，您都可以根據(jù)API的需求選擇合適的方法。遵循這些最佳實(shí)踐和策略，您可以確保長(zhǎng)時(shí)間運(yùn)行的任務(wù)得到有效管理，而不會(huì)損害用戶體驗(yàn)或系統(tǒng)性能。
閱讀
英文版
AI好書推薦
AI日新月異，但是萬(wàn)丈高樓拔地起，離不開(kāi)良好的基礎(chǔ)。您是否有興趣了解人工智能的原理和實(shí)踐？ 不要再觀望！ 我們關(guān)于 AI 原則和實(shí)踐的書是任何想要深入了解 AI 世界的人的完美資源。 由該領(lǐng)域的領(lǐng)先專家撰寫，這本綜合指南涵蓋了從機(jī)器學(xué)習(xí)的基礎(chǔ)知識(shí)到構(gòu)建智能系統(tǒng)的高級(jí)技術(shù)的所有內(nèi)容。 無(wú)論您是初學(xué)者還是經(jīng)驗(yàn)豐富的 AI 從業(yè)者，本書都能滿足您的需求。 那為什么還要等呢？
人工智能原理與實(shí)踐 全面涵蓋人工智能和數(shù)據(jù)科學(xué)各個(gè)重要體系經(jīng)典文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-686405.html
北大出版社，人工智能原理與實(shí)踐 人工智能和數(shù)據(jù)科學(xué)從入門到精通 詳解機(jī)器學(xué)習(xí)深度學(xué)習(xí)算法原理 文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-686405.html
到了這里，關(guān)于開(kāi)發(fā)API？ FastAPI有效處理長(zhǎng)時(shí)間運(yùn)行任務(wù)的策略!的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
