asins参数含180+元素时触发asyncio.exceptions.TimeoutError的解决求助

以下示例仅向事件循环中添加了一个元素（对应变量asins）。
当asins参数包含180个或更多元素时，会遇到asyncio.exceptions.TimeoutError错误。
如果用这180个元素中的任意一个创建列表，都能得到成功响应，说明问题与API无关。求解决方法。

原代码：

import asyncio
import aiohttp
import sys
import pandas as pd

def create_params(asins_set):
    params = []
    for asin in asins_set:
        param = {
            'api_key': '...',
            'type': 'product',
            'amazon_domain': 'amazon.com',
            'asin': asin,
        }
        params.append(param)
    return params

if sys.version_info[0] == 3 and sys.version_info[1] >= 8 and sys.platform.startswith('win'):
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

# creates a list of tasks to add to the event loop at once
def get_tasks(session, params):
    tasks = []
    for param in params:
        tasks.append(session.get(
            'https://api.rainforestapi.com/request',
            params = param
        ))
    return tasks

results = []
async def get_suggested(params):
    async with aiohttp.ClientSession() as session:
        tasks = get_tasks(session, params)
        responses = await asyncio.gather(*tasks)
        for response in responses:
            results.append(await response.json())
        return results

def get_asin_titles(asins_set):
    params = create_params(asins_set)
    r = asyncio.run(get_suggested(params))
    asins_and_titles = dict()
    for result in r:
        if result['request_info']['success'] == True:
            asin = result['request_parameters']['asin']
            title = result['product']['title']
            asins_and_titles[asin] = title
    return asins_and_titles

asins = ['b07wp7q5bf']

final = get_asin_titles(asins)
print(final)

1. 限制并发请求数 #

一次性发起180个请求会占满连接池，或者触发API限流机制，用asyncio.Semaphore控制同时发起的请求数量，比如限制为20个：

修改get_suggested函数如下：

async def get_suggested(params, max_concurrent=20):
    # 限制并发数
    semaphore = asyncio.Semaphore(max_concurrent)
    results = []  # 把结果列表移到函数内部，避免全局变量冲突
    
    async def fetch(session, param):
        async with semaphore:
            try:
                async with session.get(
                    'https://api.rainforestapi.com/request',
                    params=param
                ) as response:
                    return await response.json()
            except Exception as e:
                print(f"请求ASIN {param['asin']} 失败: {str(e)}")
                return None
    
    async with aiohttp.ClientSession() as session:
        # 生成所有任务
        tasks = [fetch(session, param) for param in params]
        # 执行任务，允许个别请求失败不影响整体
        responses = await asyncio.gather(*tasks)
        # 过滤掉失败的请求结果
        results = [resp for resp in responses if resp is not None]
        return results

2. 增加超时时间 #

默认超时时间可能不足以支撑大量并发请求的排队和响应，给请求设置更长的超时：

async with session.get(
    'https://api.rainforestapi.com/request',
    params=param,
    timeout=aiohttp.ClientTimeout(total=60)  # 设置总超时为60秒
) as response:

3. 适配API速率限制 #

即使单个请求没问题，批量请求可能触发API的速率限制（比如每分钟请求数上限），可以查看API文档确认允许的速率，调整并发数即可。

4. 优化全局变量问题 #

原代码中results是全局变量，在并发场景下容易出现数据冲突，改成函数内部变量更安全，上述示例已完成该调整。

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