假設有一些 API 已經在生產中運行，并且您創建了另一個 API，您有點想使用到達生產 API 的傳入請求進行 A/B 測驗。現在我想知道，是否有可能做這樣的事情，（我知道人們通過為 A/B 測驗等保留兩個不同的 API 版本來進行流量拆分）
一旦收到生產 API 的傳入請求，就向新 API 發出異步請求，然后繼續執行生產 API 的其余代碼，然后在將最終回應回傳給呼叫者之前回來，您檢查是否為之前創建的異步任務計算了結果。如果它可用，則回傳它而不是當前的 API。
我想知道，做這樣的事情的最佳方法是什么？我們是否嘗試為此撰寫裝飾器或其他東西？我有點擔心如果我們在這里使用 async 可能會發生很多邊緣情況。任何人都有關于使代碼或整個方法更好的任何指示？
謝謝你的時間！
上述方法的一些偽代碼，
import asyncio
def call_old_api():
pass
async def call_new_api():
pass
async def main():
task = asyncio.Task(call_new_api())
oldResp = call_old_api()
resp = await task
if task.done():
return resp
else:
task.cancel() # maybe
return oldResp
asyncio.run(main())
uj5u.com熱心網友回復：
你不能只call_old_api()在 asyncio 的協程中執行。有詳細的解釋，為什么在這里。請確保你了解它，因為這取決于你的服務器是如何作業的，你可能不能夠做你想做的（到同步服務器保存撰寫異步代碼的點上運行的異步API，例如）。
如果您了解自己在做什么，并且您有一個異步服務器，您可以在執行緒中呼叫舊的同步 API 并使用一個任務來運行新的 API：
task = asyncio.Task(call_new_api())
oldResp = await in_thread(call_old_api())
if task.done():
return task.result() # here you should keep in mind that task.result() may raise exception if the new api request failed, but that's probably ok for you
else:
task.cancel() # yes, but you should take care of the cancelling, see - https://stackoverflow.com/a/43810272/1113207
return oldResp
我認為您可以更進一步，而不是總是等待舊 API 完成，您可以同時運行兩個 API 并回傳完成的第一個（以防新 API 比舊 API 作業得更快）。有了上面的所有檢查和建議，它應該看起來像這樣：
import asyncio
import random
import time
from contextlib import suppress
def call_old_api():
time.sleep(random.randint(0, 2))
return "OLD"
async def call_new_api():
await asyncio.sleep(random.randint(0, 2))
return "NEW"
async def in_thread(func):
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, func)
async def ensure_cancelled(task):
task.cancel()
with suppress(asyncio.CancelledError):
await task
async def main():
old_api_task = asyncio.Task(in_thread(call_old_api))
new_api_task = asyncio.Task(call_new_api())
done, pending = await asyncio.wait(
[old_api_task, new_api_task], return_when=asyncio.FIRST_COMPLETED
)
if pending:
for task in pending:
await ensure_cancelled(task)
finished_task = done.pop()
res = finished_task.result()
print(res)
asyncio.run(main())
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/404457.html
標籤：
上一篇：在需要時解決承諾