我正在開發(fā)一個(gè) Flask 應(yīng)用程序，其中對(duì)客戶端的響應(yīng)取決于我從幾個(gè)外部 API 獲得的回復(fù)。對(duì)這些 API 的請(qǐng)求在邏輯上是相互獨(dú)立的，因此可以通過并行發(fā)送這些請(qǐng)求來實(shí)現(xiàn)速度增益（在下面的示例中，響應(yīng)時(shí)間將幾乎減少一半）。在我看來，實(shí)現(xiàn)這一點(diǎn)的最簡(jiǎn)單、最現(xiàn)代的方法是使用 asyncio 并在一個(gè)單獨(dú)的異步函數(shù)中處理所有工作，該函數(shù)是使用 asyncio.run() 從 Flask 視圖函數(shù)調(diào)用的。我在下面提供了一個(gè)簡(jiǎn)短的工作示例。將 celery 或任何其他類型的隊(duì)列與單獨(dú)的工作進(jìn)程一起使用在這里實(shí)際上沒有意義，因?yàn)轫憫?yīng)在發(fā)送回復(fù)之前必須等待 API 結(jié)果。據(jù)我所知，這是這個(gè)想法的一個(gè)變體，其中通過 asyncio 訪問處理循環(huán)。當(dāng)然有這方面的應(yīng)用程序，但我認(rèn)為如果我們真的只想在響應(yīng)請(qǐng)求之前并行化 IO，那么這就不必要地復(fù)雜了。然而，我知道在 Flask 中使用各種多線程可能存在一些陷阱。因此我的問題是：在生產(chǎn)環(huán)境中使用時(shí)，下面的實(shí)現(xiàn)是否被認(rèn)為是安全的？這與我們運(yùn)行 Flask 的服務(wù)器類型有何關(guān)系？特別是，內(nèi)置開發(fā)服務(wù)器或典型的多工作程序 Gunicorn 設(shè)置。關(guān)于異步函數(shù)中 Flask 的應(yīng)用程序和請(qǐng)求上下文是否需要考慮任何因素，或者我可以像在任何其他函數(shù)中一樣使用它們嗎？即我可以簡(jiǎn)單地導(dǎo)入 current_app 來訪問我的應(yīng)用程序配置或使用 g 和 session 對(duì)象嗎？當(dāng)寫信給他們時(shí)，顯然必須考慮可能的競(jìng)爭(zhēng)條件，但是還有其他問題嗎？在我的基本測(cè)試（不在示例中）中，一切似乎都工作正常。還有其他解決方案可以對(duì)此進(jìn)行改進(jìn)嗎？這是我的示例應(yīng)用程序。由于 ascynio 接口隨著時(shí)間的推移發(fā)生了一些變化，因此可能值得注意的是，我在 Python 3.7 和 3.8 上對(duì)此進(jìn)行了測(cè)試，并且我已盡力避免 asyncio 中已棄用的部分。import asyncioimport randomimport timefrom flask import Flaskapp = Flask(__name__)async def contact_api_a():? ? print(f'{time.perf_counter()}: Start request 1')? ? # This sleep simulates querying and having to wait for an external API? ? await asyncio.sleep(2)? ? # Here is our simulated API reply? ? result = random.random()? ? print(f'{time.perf_counter()}: Finish request 1')? ? return resultasync def contact_api_b():? ? print(f'{time.perf_counter()}: Start request 2')? ? await asyncio.sleep(1)? ? result = random.random()? ? print(f'{time.perf_counter()}: Finish request 2')? ? return resultasync def contact_apis():? ? # Create the two tasks? ? task_a = asyncio.create_task(contact_api_a())? ? task_b = asyncio.create_task(contact_api_b())? ? # Wait for both API requests to finish? ? result_a, result_b = await asyncio.gather(task_a, task_b)? ? print(f'{time.perf_counter()}: Finish both requests')
1 回答
慕絲7291255
TA貢獻(xiàn)1859條經(jīng)驗(yàn) 獲得超6個(gè)贊
這可以安全地在生產(chǎn)環(huán)境中運(yùn)行，但 asyncio 無(wú)法與 Gunicorn 異步工作線程（例如 gevent 或 eventlet）有效配合。這是因?yàn)?code>result_a, result_b = asyncio.run(contact_apis())將會(huì)阻止 gevent/eventlet 事件循環(huán)直到其完成，而使用 gevent/eventlet 生成等效項(xiàng)則不會(huì)。Flask 服務(wù)器不應(yīng)該在生產(chǎn)中使用。Gunicorn 線程工作線程（或多個(gè) Gunicorn 進(jìn)程）會(huì)很好，因?yàn)?asyncio 會(huì)阻塞線程/進(jìn)程。
全局變量可以正常工作，因?yàn)樗鼈兣c線程（線程工作線程）或綠色線程（gevent/eventlet）相關(guān)聯(lián)，而不是與 asyncio 任務(wù)相關(guān)聯(lián)。
我想說Quart是一種改進(jìn)。Quart 是使用 asyncio 重新實(shí)現(xiàn)的 Flask API。對(duì)于 Quart，上面的代碼片段是：
import asyncio
import random
import time
from quart import Quart
? ??
app = Quart(__name__)
? ??
async def contact_api_a():
? ? print(f'{time.perf_counter()}: Start request 1')
? ? # This sleep simulates querying and having to wait for an external API
? ? await asyncio.sleep(2)
? ? # Here is our simulated API reply
? ? result = random.random()
? ? print(f'{time.perf_counter()}: Finish request 1')
? ? return result
? ??
async def contact_api_b():
? ? print(f'{time.perf_counter()}: Start request 2')
? ? await asyncio.sleep(1)
? ? result = random.random()
? ? print(f'{time.perf_counter()}: Finish request 2')
? ? return result
? ??
async def contact_apis():
? ? # Create the two tasks
? ? task_a = asyncio.create_task(contact_api_a())
? ? task_b = asyncio.create_task(contact_api_b())
? ? # Wait for both API requests to finish
? ? result_a, result_b = await asyncio.gather(task_a, task_b)
? ? print(f'{time.perf_counter()}: Finish both requests')
? ? return result_a, result_b
? ??
@app.route('/')
async def hello_world():
? ? start_time = time.perf_counter()
? ? # All async processes are organized in a separate function
? ? result_a, result_b = await contact_apis()
? ? # We implement some final business logic before finishing the request
? ? final_result = result_a + result_b
? ? processing_time = time.perf_counter() - start_time
? ? return f'Result: {final_result:.2f}; Processing time: {processing_time:.2f}'
我還建議使用基于 asyncio 的請(qǐng)求庫(kù)，例如httpx
添加回答
舉報(bào)
0/150
提交
取消