aiohttp并发请求超25次触发ClientConnectorError问题求助

我需要持续发送请求直到目标状态出现，但当并发连接数超过25次时，会触发aiohttp.client_exceptions.ClientConnectorError错误。已经尝试设置ssl=False、使用sslcontext，以及调整aiohttp.TCPConnector的limit和limit_per_host参数，都没解决问题，并发低于25次时能正常运行。

我的代码 #

import aiohttp
import asyncio
import sys

if sys.platform == 'win32':
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())


async def proccess_status(headers, payload):
    await asyncio.sleep(0.3)

    async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(limit=10000, limit_per_host=10000), trust_env=True) as session:
        while True:
            async with session.post("https://example.com", headers=headers, data=payload) as response:
                try:
                    #print(response)
                    pass
                
                except Exception as error:
                    print(error)


async def main():
    number_of_requests = 50
    timeout = 18000  # Timelimit in seconds
    headers = {}
    payload = "random_payload"

    tasks = [asyncio.create_task(proccess_status(headers, payload)) for _ in range(number_of_requests)]
    
    results, pending_tasks = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED, timeout=timeout)
    
    if results:
        first_task_done = results.pop()
        value = first_task_done.result()
    else:
        print("No requests were completed within the time limit.")
    
    for task in pending_tasks:
        task.cancel()

    return value


if __name__ == "__main__":
    asyncio.run(main())

预期结果 #

<ClientResponse(https://example.com) [200 OK]>
<CIMultiDictProxy('Connection': 'close', 'Content-Length': '371', 'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*', 'Content-Encoding': 'br', 'Date': 'Sun, 21 Jan 2024 17:02:57 GMT', 'Timing-Allow-Origin': '*')>

实际错误 #

aiohttp.client_exceptions.ClientConnectorError: Cannot connect to host example.com:port ssl:default [Connect call failed ('IP', PORT)]

解决方案分析 #

你遇到的问题核心是目标服务器的并发连接限制，而非aiohttp的配置问题。多数服务器会对单IP来源的并发连接数做严格限制（比如你遇到的25次阈值），超过后直接拒绝新连接，这就是报错的根本原因。

具体调整建议： #

1. 复用ClientSession
   你的代码每个任务都创建独立的ClientSession，相当于每个任务都开了一个新连接池，完全没必要。应该全局创建一个ClientSession，让所有任务复用同一个连接池，既能有效控制连接数，还能复用已建立的连接。
   修改后的核心代码示例：

   async def proccess_status(session, headers, payload):
       await asyncio.sleep(0.3)
       while True:
           try:
               async with session.post("https://example.com", headers=headers, data=payload) as response:
                   # 这里替换成你的目标状态判断逻辑，满足条件就返回结束任务
                   if response.status == 200:
                       return await response.json()
           except Exception as error:
               print(error)
               await asyncio.sleep(1)  # 出错后短暂休眠再重试，避免狂轰服务器
   
   async def main():
       number_of_requests = 50
       timeout = 18000
       headers = {}
       payload = "random_payload"
   
       # 全局创建一个ClientSession，所有任务复用
       async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(limit=20, limit_per_host=20), trust_env=True) as session:
           tasks = [asyncio.create_task(proccess_status(session, headers, payload)) for _ in range(number_of_requests)]
           results, pending_tasks = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED, timeout=timeout)
           
           # 后续逻辑保持不变...
   

2. 合理设置连接池上限
   既然服务器限制25次并发，就把limit和limit_per_host设为低于25的值（比如20），主动避开服务器的限制阈值，不要盲目设成超大数字。

3. 添加错误重试机制
   遇到连接错误时，不要直接抛出，短暂休眠后再重试，避免瞬间大量请求触发服务器的防护机制。

4. 补全任务结束逻辑
   你的proccess_status函数没有返回值，导致first_task_done.result()会拿到None，必须在函数里添加目标状态的判断逻辑，满足条件时返回结果，任务才会正常结束。

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