使用aiohttp.ClientSession发异步POST请求触发asyncio.TimeoutError求解

• 未显式配置aiohttp超时规则：aiohttp.ClientSession默认总超时为300秒，单批次3万请求如果遇到服务端限流、网络波动，整体耗时超过阈值就会触发asyncio.TimeoutError。
• 无并发限流逻辑：单批次3万请求直接全部生成异步任务提交，瞬间并发过高会导致本地端口耗尽、TCP连接队列拥堵，或者触发目标接口限流，大量请求等待时间过长触发超时。
• 重试逻辑未覆盖超时异常：当前backoff装饰器只捕获了连接类异常，TimeoutError不在捕获列表中，超时请求不会触发重试直接抛出。
• asyncio.gather默认异常终止机制：任意一个任务抛出未捕获的异常，gather会直接终止整批任务的等待，抛出异常导致整批数据处理失败。

1. 显式配置超时规则 #

创建ClientSession时单独配置单请求超时，禁用全局总超时，避免整批超时被一刀切：

# 创建session时指定超时
timeout = aiohttp.ClientTimeout(total=None, sock_connect=10, sock_read=30)
async with aiohttp.ClientSession(timeout=timeout) as session:

参数说明：

• total=None 禁用整session的总超时限制
• sock_connect=10 单个请求TCP连接超时时间10秒
• sock_read=30 单个请求响应读取超时时间30秒

2. 添加并发限流 #

用asyncio.Semaphore控制同时发起的请求数，根据目标接口的承受能力调整并发阈值，一般设置为100~500即可：

# 并发限制初始化，放在concurr_req方法内
semaphore = asyncio.Semaphore(200)

# 修改get_responses方法，加锁控制并发
async def get_responses(self, session, data):
    async with semaphore:
        # 原有逻辑不变

3. 扩展重试异常捕获范围 #

把超时异常加入backoff的捕获列表：

@backoff.on_exception(backoff.expo,
                      (aiohttp.ClientResponseError, aiohttp.ServerDisconnectedError, 
                       aiohttp.ClientOSError, asyncio.TimeoutError),
                      max_tries=10)

4. 优化gather异常处理 #

给asyncio.gather添加return_exceptions=True参数，单个任务异常不会终止整批任务，后续可统一过滤处理失败的请求：

results = await asyncio.gather(*tasks, return_exceptions=True)
# 后续可以遍历results判断是否为异常类型，单独处理失败请求

5. 可选优化 #

不建议用全局变量存储请求结果，可改为每个任务返回结果，最后统一汇总，避免多协程操作全局变量的潜在问题。

内容的提问来源于stack exchange，提问作者Kulasangar