使用aiohttp发起大量GET请求时如何限制并发避免文件打开数超限

核心问题排查 #

• 你在api_call函数内部为每个请求单独创建ClientSession和TCPConnector，设置的连接数限制只作用于单个请求对应的连接池，完全没有全局限制效果，8000个请求就会创建8000个独立连接，触发系统打开文件数上限报错
• 一次性创建全部8000个请求任务，没有做全局并发控制，所有请求会同时发起

具体修复方法 #

推荐用asyncio.Semaphore做全局并发控制，同时全局复用同一个ClientSession实例，修改后的代码如下：

import asyncio
from aiohttp import ClientSession, TCPConnector

# 全局配置：并发数限制设为20，可根据需求调整
CONCURRENCY_LIMIT = 20
requests_headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:93.0) Gecko/20100101 Firefox/93.0",
    "Upgrade-Insecure-Requests": "1",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.5", 
    "Accept-Encoding": "gzip, deflate, br"
}

# 全局信号量控制并发数
semaphore = asyncio.Semaphore(CONCURRENCY_LIMIT)

async def api_call(endpoint, session):
    # 先获取信号量，超过并发限制的请求会在这里等待
    async with semaphore:
        try:
            async with session.get(endpoint) as response:
                assert response.status == 200
                return await response.json()
        except Exception as e:
            self.logger.error(f"Error retrieving data: {e}")
            return None

async def main():
    # 全局复用同一个Connector和ClientSession，连接数限制和信号量保持一致
    connector = TCPConnector(limit=CONCURRENCY_LIMIT, limit_per_host=CONCURRENCY_LIMIT)
    async with ClientSession(headers=requests_headers, connector=connector) as session:
        # 替换为你实际的8000个endpoint列表
        endpoints = ["your_api_endpoint"] * 8000
        tasks = [asyncio.create_task(api_call(ep, session)) for ep in endpoints]
        # 处理返回结果
        for coro in asyncio.as_completed(tasks):
            result = await coro
            # 这里写你自己的结果处理逻辑
            if result:
                pass

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

补充说明 #

• 你可以调整CONCURRENCY_LIMIT的数值修改并发上限，建议根据目标接口的限流规则和自身系统负载调整，一般单域名并发不要超过50
• 如果需要更精细的任务调度逻辑，也可以用asyncio.Queue实现生产者消费者模式来控制并发

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