如何批量调用REST API？处理10万条Excel/CSV中的name参数

优先验证：API是否支持批量请求 #

这是最高效的解决方案，先确认目标API是否接受批量参数。比如能不能把请求载荷改成：

{
  "names": ["name1", "name2", "name3"],
  "something": "all rest values are fixed"
}

如果后端支持这种批量格式，你可以把Excel/CSV里的name按批次打包（比如一次传50-100个），调用次数直接从10万降到2000次以内，耗时能大幅压缩。

当API仅支持单条调用时的实现步骤 #

如果API只能处理单条name请求，那就通过并发请求来提升效率，具体步骤如下：

1. 读取并预处理数据 #

用工具快速读取Excel/CSV里的name值，同时过滤空值、无效值：

• Python可以用pandas或csv模块：df = pd.read_csv('your_file.csv'); names = df['name'].dropna().tolist()
• 其他语言比如Java用Apache POI，Node.js用csv-parser，核心都是把name提取成一个干净的列表。

2. 实现并发请求控制 #

因为单次调用耗时0.8-1秒，串行调用要27小时以上，必须用并发，但要控制并发数避免触发API限流：

• 异步IO（推荐）：Python用aiohttp，Node.js用axios结合async/await，这类框架适合IO密集型的API调用，资源占用比多线程低。
• 并发数设置：先从小值测试（比如10-20），如果没有报错再逐步提升（最高别超过50，除非API明确说明支持更高并发）。

3. 错误处理与重试 #

API调用难免失败，必须做容错处理：

• 捕获超时、5xx等临时错误，用指数退避重试（比如第一次等1秒，第二次2秒，第三次4秒），避免频繁重试加重服务器负担。
• 记录所有失败的name，后续单独处理，避免遗漏数据。

4. 进度监控与结果留存 #

• 实时打印已完成的请求数或批次，方便掌握进度。
• 把成功/失败的结果写入文件或数据库，避免重复调用，也方便后续核对。

示例代码（Python + aiohttp） #

import aiohttp
import asyncio
import pandas as pd

# 配置参数
API_ENDPOINT = "https://your-api-url.com"
MAX_CONCURRENT = 30  # 并发数，可根据API限流调整
RETRY_LIMIT = 3       # 最大重试次数

# 读取并清洗数据
def load_names(file_path):
    df = pd.read_csv(file_path)
    return df['name'].dropna().str.strip().tolist()

# 单条API调用逻辑（含重试）
async def single_api_call(session, name):
    payload = {
        "name": name,
        "something": "all rest values are fixed"
    }
    for attempt in range(RETRY_LIMIT):
        try:
            async with session.post(API_ENDPOINT, json=payload) as resp:
                if resp.status in (200, 201):
                    return {"name": name, "status": "success"}
                else:
                    print(f"[{attempt+1}/{RETRY_LIMIT}] {name} 调用失败，状态码: {resp.status}")
        except Exception as e:
            print(f"[{attempt+1}/{RETRY_LIMIT}] {name} 调用出错: {str(e)}")
        await asyncio.sleep(2 ** attempt)  # 指数退避等待
    return {"name": name, "status": "failed"}

# 批量执行逻辑
async def main():
    names = load_names("names.csv")
    async with aiohttp.ClientSession() as session:
        # 分批创建任务，控制并发数
        for start in range(0, len(names), MAX_CONCURRENT):
            end = min(start + MAX_CONCURRENT, len(names))
            batch_names = names[start:end]
            tasks = [single_api_call(session, name) for name in batch_names]
            results = await asyncio.gather(*tasks)
            
            # 保存当前批次结果
            with open("api_results.csv", "a", encoding="utf-8") as f:
                for res in results:
                    f.write(f"{res['name']},{res['status']}\n")
            
            print(f"已完成 {end}/{len(names)} 条请求")

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

关键注意事项 #

• 限流合规：务必遵守API服务商的限流规则，否则可能被封禁IP或暂停服务。
• 资源控制：如果是超大规模数据，避免一次性把所有name加载到内存（不过10万条字符串内存占用很小，一般没问题）。
• 幂等性：确认API是幂等的（多次调用同一name不会产生重复数据），否则重试可能导致数据重复。

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