在 Python 中使用 asyncio 进行 a/b 测试

Question

假设有一些 API 已经在生产环境中运行，并且您创建了另一个 API，您希望使用到达生产 API 的传入请求进行 A/B 测试。现在我想知道，是否有可能做这样的事情，（我知道人们通过为 A/B 测试等保留两个不同的 API 版本来进行流量拆分）

收到生产 API 的传入请求后，您立即向新 API 发出异步请求，然后继续处理生产 API 的其余代码，然后在返回最终响应之前返回给调用者，您检查是否为您之前创建的异步任务计算了结果。如果它可用，则返回它而不是当前 API。

我想知道，做这样的事情最好的方法是什么？我们是否尝试为此或其他东西编写装饰器？如果我们在这里使用异步，我有点担心会发生很多边缘情况。任何人有任何关于使代码或整个方法更好的建议吗？

感谢您的宝贵时间！

---

上述方法的一些伪代码，

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())

Answer 1

你不能只在 asyncio 的协程中执行 call_old_api()。有详细解释为什么here。请确保您理解它，因为根据您的服务器的工作方式，您可能无法做您想做的事情（在同步服务器上运行异步 API，保留编写异步代码的点，for example）。

如果你明白你在做什么，并且你有一个异步服务器，你可以在线程中调用旧的同步 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 比旧 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())