快捷導(dǎo)航

fastapi框架異步執(zhí)行踩坑記錄

更新時(shí)間：2024年06月21日 14:38:40 作者：余生的觀瀾

這篇文章主要介紹了fastapi框架異步執(zhí)行踩坑記錄,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

fastapi框架異步執(zhí)行踩坑

問(wèn)題與背景

在公司項(xiàng)目中，使用了python作為中間件實(shí)現(xiàn)的主語(yǔ)言，項(xiàng)目是中間件性質(zhì)的需要并行的為第三方提供服務(wù)，但是在使用過(guò)程中，出現(xiàn)了并行的問(wèn)題，多個(gè)耗時(shí)操作，在串行，導(dǎo)致項(xiàng)目的分析操作耗時(shí)巨大。

web技術(shù)采用的是fastapi，這個(gè)技術(shù)官網(wǎng)介紹是并行的，用java起多線程訪問(wèn)接口，得到的結(jié)論竟然是串行的，這地方就特意研究了一下。

最佳實(shí)踐

這篇文章，給我答疑解惑了：http://www.dbjr.com.cn/python/323052zd6.htm

我在項(xiàng)目中，翻了一個(gè)使用的錯(cuò)誤，類似的方式如下：

 @router.get("/a")
async def a():
    time.sleep(1)
    return {"message": "異步模式，但是同步執(zhí)行sleep函數(shù)，執(zhí)行過(guò)程是串行的"}

這種寫(xiě)法在于接口訪問(wèn)是異步的，但是執(zhí)行的時(shí)候卻是在主線程中，用同步的方法，就導(dǎo)致了串行了。

所以async和await應(yīng)該是成對(duì)出現(xiàn)的，否則就不是絕對(duì)的異步：

import asyncio

@router.get("/b")
async def b():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, time.sleep, 1)
    return {"message": "線程池中運(yùn)行sleep函數(shù)"}

這種方案接口可以異步調(diào)用，并且接口中的內(nèi)容也會(huì)被異步執(zhí)行，在實(shí)際使用中，可以對(duì)方法中的內(nèi)容進(jìn)行封裝，然后通過(guò)loop.run_in_executor去異步調(diào)用封裝的一個(gè)方法，最終實(shí)現(xiàn)絕對(duì)的異步。

fastapi的異步大騙/局

有部分入坑fastapi的同志們，可能經(jīng)常會(huì)遇到一個(gè)耗時(shí)的密集型IO任務(wù)阻塞主線程的情況，導(dǎo)致接口單個(gè)接口占用整體的線程時(shí)間過(guò)長(zhǎng)，從而影響到其他的接口響應(yīng)。

很多人可能會(huì)說(shuō)，fastapi不是一個(gè)異步的框架嗎？為什么會(huì)存在阻塞這個(gè)問(wèn)題？

這是一個(gè)很好的問(wèn)題，我之前也想過(guò)這個(gè)問(wèn)題，但是后面才發(fā)現(xiàn)，原來(lái)是fastapi框架的接口，本身不會(huì)自動(dòng)幫忙處理長(zhǎng)耗時(shí)的任務(wù)，如生成式的請(qǐng)求，調(diào)用復(fù)雜的第三方api，這個(gè)過(guò)程大概需要10-20s。

假如說(shuō)自己使用的http請(qǐng)求是直接requests庫(kù)的請(qǐng)求，那么就會(huì)導(dǎo)致阻塞。

正確的做法應(yīng)該是：

使用一個(gè)異步執(zhí)行的請(qǐng)求方式，例如aiohttp庫(kù)，手動(dòng)為請(qǐng)求添加上這個(gè)異步的請(qǐng)求方法，我就是使用這個(gè)庫(kù)解決了長(zhǎng)耗時(shí)任務(wù)阻塞線程的問(wèn)題。

首先得安裝好這個(gè)庫(kù)：

pip install aiohttp

然后就再代碼當(dāng)中

from fastapi import FastAPI
import aiohttp
import asyncio

app = FastAPI()

async def long_running_task(url: str):
    async with aiohttp.ClientSession() as session:
        async with session.post(url, json={"load": "data"}) as response:
            # 處理響應(yīng)數(shù)據(jù)
            data = await response.json()
            return data

@app.post("/start_task/")
async def start_task():
    url = "http://example.com/api"  # 目標(biāo) API 的 URL
    task_result = await long_running_task(url)
    return task_result

這樣，這個(gè)長(zhǎng)任務(wù)就不會(huì)阻塞主線程了。