python - 异步 http 调用的时间是应有的两倍

Question

我正在学习异步和信号量，我已经使用 fastapi 制作了一个端点来进行测试。 fastapi 端点只是一个简单的服务器端，它接受一个带有休眠时间的请求，并在返回响应之前休眠很长时间。我通过 uvicorn 运行 fastapi 以进行测试，但有 5 个工作人员。这仅用于测试，我了解生产我应该使用 gunicorn 和 nginx，但出于学习的目的，我只是使用 uvicornuvicorn api_example:app --port 8008 --host cdsre.co.uk --workers 5 api_example.py

from time import sleep, time
import json
from fastapi import FastAPI, Response, Request


app = FastAPI()


@app.post("/sleeper")
async def sleeper(request: Request):
    request_data = await request.json()
    sleep_time = request_data['sleep_time']
    start = time()
    sleep(sleep_time)
    end = time()
    return Response(content=json.dumps({"slept for": end - start}))

我本地机器上的客户端代码正在尝试利用异步和信号量来并行调用 3 个发布请求。我有 6 个请求， sleep 定时器为 5 秒。所以这里的期望是处理这 6 个请求大约需要 10 秒。
async_example.py

import aiohttp
import asyncio
import time


async def get_http_response(session, url):
    async with semaphore:
        print("firing request...")
        start = time.time()
        async with session.post(url, json={"sleep_time": 5}) as resp:
            response = await resp.text()
            end = time.time()
            print(f"Client time: {end - start}, server time: {response}")
            return response


async def main():
    async with aiohttp.ClientSession() as session:
        tasks = []
        for number in range(6):
            url = f'http://cdsre.co.uk:8008/sleeper'
            tasks.append(asyncio.ensure_future(get_http_response(session, url)))
        responses = await asyncio.gather(*tasks)
        for response in responses:
            pass
            # print(response)


semaphore = asyncio.Semaphore(3)
start_time = time.time()
asyncio.get_event_loop().run_until_complete(main())
print("--- %s seconds ---" % (time.time() - start_time))

然而，发布请求通常需要两倍的时间。在这种情况下， sleep 定时器为 5 秒，一些请求需要 10 秒。

firing request...
firing request...
firing request...
Client time: 5.137275695800781, server time: {"slept for": 5.005105018615723}
firing request...
Client time: 10.158655643463135, server time: {"slept for": 5.0042970180511475}
Client time: 10.158655643463135, server time: {"slept for": 5.001959800720215}
firing request...
firing request...
Client time: 5.055504560470581, server time: {"slept for": 5.005110025405884}
Client time: 5.056135654449463, server time: {"slept for": 5.005115509033203}
Client time: 5.107320070266724, server time: {"slept for": 5.005107402801514}
--- 15.271023750305176 seconds ---

有时它慢 3 倍，它总是与我的 sleep 时间有关，这让我觉得发生了某种排队或我错过了某种竞争条件，但我认为信号量模式的全部目的是避免这些竞争条件，这样我在任何时候限制为 3 个请求总是会少于服务器端可用的工作(5 个工作人员)，因此应该始终有一个可用的服务器端来处理它。
我也不会在信号量内部开始计时，所以我不会提前启动它，所以它应该只在发送请求时启动计时器。希望我只是错过了一些明显的东西。如果有人想尝试，我已经留下了终点网址。我将不胜感激任何帮助解决这个问题。本质上，我需要能够编写一个异步客户端，该客户端可以并行发送请求到一个限制，并且在测量响应时间时保持一致。
一些耗时 3 倍的示例

firing request...
firing request...
firing request...
Client time: 15.127191305160522, server time: {"slept for": 5.001192808151245}
firing request...
Client time: 15.127155303955078, server time: {"slept for": 5.005094766616821}
Client time: 15.127155303955078, server time: {"slept for": 5.005074977874756}
firing request...
firing request...
Client time: 5.053789854049683, server time: {"slept for": 5.005076169967651}
Client time: 5.100871801376343, server time: {"slept for": 5.005076885223389}
Client time: 10.107984781265259, server time: {"slept for": 5.005110502243042}
--- 25.236175775527954 seconds ---

Answer 1

继续评论:

You’re using time.sleep. This will block the thread. You want to useawait asyncio.sleep(sleep_time). - @dirn (this is the answer)

However, having taken your comment and applied it in my code theresults are a lot more consistent . So what is the difference betweentime.sleep in a worker and await asyncio.sleep.......I was thinkingthat each worker would get a separate request, so a sleep blockingthat thread wouldn't affect the other - . @Chris Doyle

虽然我没有检查 uvunicorn 默认是如何设置它的工作人员的，但使用异步进行编码的重点是在同一个线程上做所有事情——或者尽可能多地做。
由于您的 View 被定义为 async尤其如此——这种编码范式不期望异步函数在等待任何调用时阻塞——线程将被停止。
如果不是 sleep您必须调用一些需要时间的代码，并且还没有准备好异步，然后您可以通过将委托(delegate)函数运行到单独的线程中来使该调用异步。 Python asyncio 通过提供 loop.run_in_executor 让这一切变得简单。 call:它会自动创建一个带有线程池的基于线程的执行器，并在一个单独的线程中运行你的阻塞函数，同时释放当前线程用于异步事件循环以进一步协调其他任务/ worker 。
否则，如前所述，如果您使用 sleep出于测试目的，请等待 asyncio.sleep反而。您可以使用 await loop.run_in_executor(None, time.sleep, 5)检查它在阻塞函数中的表现。