python - 如何在 ClientOSError : [Errno 104] Connection reset by peer? 时重试异步请求

Question

我在 Google Cloud 中有一个函数可以接受多个参数。我使用 aiohttp 生成具有不同参数值组合的 ~2k 异步请求:

# url = 'https://...'
# headers = {'X-Header': 'value'}

timeout = aiohttp.ClientTimeout(total=72000000)

async def submit_bt(session, url, payload):
        async with session.post(url, json=payload) as resp:
                result = await resp.text()

async def main():
        async with aiohttp.ClientSession(headers=headers, timeout=timeout) as session:
                tasks = []
                gen = payload_generator() # a class that generates dictionaries
                for payload in gen.param_grid():
                        tasks.append(asyncio.ensure_future(submit_bt(session, url, payload)))

                bt_results = await asyncio.gather(*tasks)
                for result in bt_results:
                        pass

asyncio.run(main())

一个函数运行需要 3 到 6 分钟，函数超时设置为 9 分钟，最大实例数为 3000，但即使提交的请求总数为在 1.5k 和 2.5k 之间。在某些情况下，所有请求都会在 20 到 30 分钟内得到处理，但有时我会在客户端收到错误消息:

ClientOSError: [Errno 104] Connection reset by peer

这不对应于服务器端的任何错误。我想我应该能够将其捕获为 aiohttp.client_exceptions.ClientOSError 异常，但我不确定如何在异步设置中处理它，以便重新提交失败的请求并提前终止被避免。非常感谢任何提示。

Answer 1

@vaizki 在评论中建议的解决方案似乎对我很有效。仔细查看回溯后发现异常是在 submit_bt 协程中引发的，所以我添加了 try-except 子句:

async def submit_bt(session, url, payload):
        try:
                async with session.post(url, json=payload) as resp:
                        result = await resp.text()
        except aiohttp.client_exceptions.ClientOSError as e:
                await asyncio.sleep(3 + random.randint(0, 9))
                async with session.post(url, json=payload) as resp:
                        result = await resp.text()
        except Exception as e:
                result = str(e)
        return result

重复的行看起来不是很优雅，但这对我来说仍在进行中，代码结构在这个阶段还没有正式化。无论如何很清楚地看到我想要实现的目标:

• 将有效负载发布到函数 URL，
• 捕获异常，但仅在 ClientOSError 的情况下重复该帖子，并且仅重复一次。

我不想使用 while True 循环来避免在出现某些严重问题时无限执行。我按原样尝试了这段代码几次，我知道它通过几次连接重置工作，直到任务列表结束，因为我得到了函数生成的所有结果，所以即使在这种形式下它也足够健壮我的情况。