Python 3.5+异步并行REST API分页请求实现难题

我明白你的需求啦——要针对同一个支持偏移分页的REST API，同时发起多个并行请求，而不是用静态的URL列表。其实这个需求比静态列表的场景更实用，咱们一步步来实现它。

本质上就是动态生成带不同offset参数的请求任务，然后用asyncio.gather()让这些任务并行执行。如果数据量很大，还可以分批发起并行请求，避免一次性给API造成过大压力。

下面是一个可直接运行的完整实现，你只需要替换成自己的API端点和参数即可：

import asyncio
import aiohttp

async def fetch_page(session: aiohttp.ClientSession, offset: int, limit: int = 10):
    """封装单个分页请求的逻辑，返回接口响应数据"""
    # 替换成你的实际API端点
    api_url = "https://your-target-api.com/your-endpoint"
    request_params = {
        "offset": offset,
        "limit": limit
    }

    try:
        # 使用aiohttp会话发起请求，自动管理连接
        async with session.get(api_url, params=request_params) as response:
            if response.status == 200:
                return await response.json()
            else:
                print(f"请求失败 | 状态码: {response.status} | 偏移量: {offset}")
                return None
    except Exception as e:
        print(f"请求出错 | 错误信息: {str(e)} | 偏移量: {offset}")
        return None

async def main():
    # 配置参数，根据你的API实际情况调整
    limit_per_page = 10       # 每页返回的数据条数
    parallel_count = 2        # 每次并行发起的请求数
    total_items = 100         # 假设API总共有100条数据（如果不知道可以先请求一次获取）

    # 生成所有需要的偏移量列表：0,10,20,...,90
    all_offsets = list(range(0, total_items, limit_per_page))

    # 创建aiohttp会话（复用连接更高效）
    async with aiohttp.ClientSession() as session:
        # 分批处理请求，避免一次性发起过多请求触发限流
        for batch_start in range(0, len(all_offsets), parallel_count):
            # 取出当前批次的偏移量
            current_batch_offsets = all_offsets[batch_start:batch_start+parallel_count]
            # 为每个偏移量创建异步任务
            tasks = [fetch_page(session, offset, limit_per_page) for offset in current_batch_offsets]
            # 并行执行所有任务，等待全部完成后获取结果
            batch_results = await asyncio.gather(*tasks)

            # 处理当前批次的结果（这里仅做示例打印，实际可替换为保存/合并逻辑）
            for offset, result in zip(current_batch_offsets, batch_results):
                if result:
                    print(f"✅ 偏移量 {offset} 数据获取成功，共 {len(result['data'])} 条")

if __name__ == "__main__":
    # 运行异步主函数
    asyncio.run(main())

1. fetch_page函数：

   • 复用aiohttp.ClientSession来管理连接池，比每次新建连接更高效
   • 内置了基础的异常和错误状态码处理，避免单个请求失败导致整个程序崩溃
   • 接收offset参数动态构造请求，完美适配分页场景

2. 分批并行逻辑：

   • 通过range和切片实现分批，比如每次取2个偏移量发起并行请求
   • 如果你的API没有严格限流，也可以去掉分批逻辑，直接一次性生成所有任务并行执行

3. 动态获取总数据量：
   如果不知道API的总数据条数，可以先发起一次请求获取：

   # 在main函数开头添加
   async with session.get(api_url, params={"offset":0, "limit":1}) as init_resp:
       init_data = await init_resp.json()
       total_items = init_data["total"]  # 假设API返回的总条数字段是total
   

• API限流：大部分公开API都有请求频率限制，并行数不要设置得太高，否则可能收到429错误，必要时可以在批次之间加await asyncio.sleep(1)做简单限流
• 重试机制：如果API偶尔会失败，可以给fetch_page添加重试逻辑（比如用tenacity库，或者自己写循环重试）
• 结果处理：示例中只是打印结果，实际场景可以把所有结果合并成一个列表，或者直接写入数据库

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