我正在開發一個 Flask 應用程序，其中對客戶端的響應取決于我從幾個外部 API 獲得的回復。對這些 API 的請求在邏輯上是相互獨立的，因此可以通過并行發送這些請求來實現速度增益（在下面的示例中，響應時間將幾乎減少一半）。在我看來，實現這一點的最簡單、最現代的方法是使用 asyncio 并在一個單獨的異步函數中處理所有工作，該函數是使用 asyncio.run() 從 Flask 視圖函數調用的。我在下面提供了一個簡短的工作示例。將 celery 或任何其他類型的隊列與單獨的工作進程一起使用在這里實際上沒有意義，因為響應在發送回復之前必須等待 API 結果。據我所知，這是這個想法的一個變體，其中通過 asyncio 訪問處理循環。當然有這方面的應用程序，但我認為如果我們真的只想在響應請求之前并行化 IO，那么這就不必要地復雜了。然而，我知道在 Flask 中使用各種多線程可能存在一些陷阱。因此我的問題是：在生產環境中使用時，下面的實現是否被認為是安全的？這與我們運行 Flask 的服務器類型有何關系？特別是，內置開發服務器或典型的多工作程序 Gunicorn 設置。關于異步函數中 Flask 的應用程序和請求上下文是否需要考慮任何因素，或者我可以像在任何其他函數中一樣使用它們嗎？即我可以簡單地導入 current_app 來訪問我的應用程序配置或使用 g 和 session 對象嗎？當寫信給他們時，顯然必須考慮可能的競爭條件，但是還有其他問題嗎？在我的基本測試（不在示例中）中，一切似乎都工作正常。還有其他解決方案可以對此進行改進嗎？這是我的示例應用程序。由于 ascynio 接口隨著時間的推移發生了一些變化，因此可能值得注意的是，我在 Python 3.7 和 3.8 上對此進行了測試，并且我已盡力避免 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貢獻1859條經驗 獲得超6個贊
這可以安全地在生產環境中運行，但 asyncio 無法與 Gunicorn 異步工作線程（例如 gevent 或 eventlet）有效配合。這是因為
result_a, result_b = asyncio.run(contact_apis())將會阻止 gevent/eventlet 事件循環直到其完成，而使用 gevent/eventlet 生成等效項則不會。Flask 服務器不應該在生產中使用。Gunicorn 線程工作線程（或多個 Gunicorn 進程）會很好，因為 asyncio 會阻塞線程/進程。全局變量可以正常工作，因為它們與線程（線程工作線程）或綠色線程（gevent/eventlet）相關聯，而不是與 asyncio 任務相關聯。
我想說Quart是一種改進。Quart 是使用 asyncio 重新實現的 Flask API。對于 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 的請求庫，例如httpx
添加回答
舉報
0/150
提交
取消