超时后获得完成的 aiohttp 并行请求

Posted 2023-03-30

【中文标题】超时后获得完成的 aiohttp 并行请求

【英文标题】：get completed aiohttp parallel requests after timeout

【发布时间】：2020-01-23 08:57:51

【问题描述】：

我正在使用aiohttp 执行一些并行的 HTTP 发布请求。

为了不超过阈值，我必须全局设置超时（在 ClientSession 上）。

问题是我想接受我在阈值之前完成的（部分会话）响应，例如，如果会话包含 10 个请求并且在超时之前我已经完成了其中的 5 个，我想取这5个的结果。但我还没有弄清楚如何做到这一点。

我使用的代码是这样的：

import aiohttp
import asyncio
import requests

async def fetch(session):
    async with session.get("https://amazon.com") as response:
        return response.status

async def main(n, timeout):
    async with aiohttp.ClientSession(timeout=timeout) as session:
        return await asyncio.gather(*(fetch(session) for _ in range(n)))

timeout = aiohttp.ClientTimeout(total=0.4)
res = asyncio.run(main(10, timeout))
print(res)

使用timeout = 0.4 会引发asyncio.TimeoutError，我不知道如何获得部分执行的响应。

例如，如果我将超时设置为 5 秒，则所有请求都已完成，我会获得一个包含十个 200 的列表。

谢谢

【参考方案1】：

使用asyncio.wait 代替asyncio.gather

另请参阅this QA，了解有关差异的更多信息。

注意：wait 的 timeout 参数以秒为单位。

最重要的是，您可能根本不需要为 ClientSession 指定超时时间。

重新编写的代码（为了增加响应时间的差异，我添加了几个不同的来源并执行了 20 个请求）

import asyncio
import random
import aiohttp
import requests

sources = ["amazon.com", "hotmail.com", "***.com"]

async def fetch(session):
    rnd = random.choice(sources)
    async with session.get(f"https://rnd") as response:
        return response.status

async def main(n, timeout):
    async with aiohttp.ClientSession() as session:
        completed, pending = await asyncio.wait(
            [fetch(session) for _ in range(n)],
            timeout=timeout
        )
    for t in pending:  # cancel the pending tasks
        t.cancel()
    return [t.result() for t in completed]

timeout = 0.5
res = asyncio.run(main(20, timeout))
print(res)

timeout 的值随着 0.3、0.5 和 0.8 的增加而增加

(.venv) async_req_timeout $ python async_req_timeout.py 
[200, 200]

(.venv) async_req_timeout $ python async_req_timeout.py 
[200, 200, 200, 200, 200, 200, 200, 200, 200, 200]

(.venv) (base) async_req_timeout $ python async_req_timeout.py 
[200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200]

【讨论】：

有了这个我有TypeError: wait() takes 1 positional argument but 10 positional arguments (and 1 keyword-only argument) were given，我想在*(fetch(session) for _ in range(n))

是的，当然，不需要解包参数：只将生成器表达式传递给它。已更正

它不起作用。对于(fetch(session) for _ in range(n))，我可以使用[fetch(session) for _ in range(n)] 纠正一个错误。如果我在asyncio.wait 中设置超时，我将获得Session is closed，将其设置在ClientSession 中有效，但如果超时，我将无法获得 200 个响应。如果它在超时之前完成，我将获得 200

的列表

这两种超时表示方式不同：一种是ClientTimeout类，一种是int/float，以秒表示，见我上面的注释。尝试在没有超时的情况下创建 ClientSession，并让超时为 int 或 float 而不是 ClientTimeout。也就是说，只在wait 上使用超时。或者在 ClientSession 上使用更长的超时时间，在 wait 上使用更短的超时时间

请看最终答案，现在包含完整代码