Python 并发请求实战：用 asyncio 把批量 API 调用提速 10 倍

在数据采集、AI 模型调用、文件处理等项目中我们经常需要批量请求 API。最直接的写法是使用requests逐个发送请求但当任务数量达到几百甚至几千时串行执行会浪费大量时间。这篇文章将从一个简单案例出发介绍如何使用 Python 的asyncio和aiohttp实现并发请求并补充超时控制、限流、重试等实际项目中不可缺少的机制。一、为什么串行请求速度慢假设我们需要请求 100 个接口每个接口平均响应时间为 500 毫秒。使用串行方式时总耗时大约为100 × 0.5 秒 50 秒传统写法如下import requests def fetch(url): response requests.get(url, timeout10) response.raise_for_status() return response.json() urls [ fhttps://example.com/api/items/{item_id} for item_id in range(100) ] results [] for url in urls: results.append(fetch(url))这段代码的问题并不在于单次请求慢而在于程序发出请求后会一直等待服务器返回结果。在等待期间CPU 几乎没有做任何事情。对于这种以网络等待为主的 I/O 密集型任务并发执行通常能够显著缩短整体耗时。二、使用 asyncio 和 aiohttp 实现并发首先安装依赖pip install aiohttp然后将串行代码改造成异步版本import asyncio import aiohttp async def fetch(session, url): async with session.get(url) as response: response.raise_for_status() return await response.json() async def main(): urls [ fhttps://example.com/api/items/{item_id} for item_id in range(100) ] async with aiohttp.ClientSession() as session: tasks [ fetch(session, url) for url in urls ] results await asyncio.gather(*tasks) print(f成功获取 {len(results)} 条数据) if __name__ __main__: asyncio.run(main())这里有三个关键点使用async def定义异步函数。使用await等待网络请求完成。使用asyncio.gather()同时调度多个任务。程序不再等待一个请求结束后才发送下一个请求而是可以在等待某个接口响应时继续处理其他请求。三、并发不是越高越好如果一次性创建几千个请求可能会遇到以下问题触发服务端限流本地连接数耗尽内存占用快速上升大量请求同时超时被目标服务判断为异常流量。因此实际项目中通常要使用信号量限制并发数量。import asyncio import aiohttp CONCURRENCY 10 semaphore asyncio.Semaphore(CONCURRENCY) async def fetch(session, url): async with semaphore: async with session.get(url) as response: response.raise_for_status() return await response.json()当CONCURRENCY设置为 10 时程序最多同时执行 10 个请求。其他任务会进入等待状态直到有可用的并发名额。并发数没有适用于所有项目的固定答案可以从 520 开始测试再根据接口限流规则、服务器性能和实际响应时间进行调整。四、为请求设置超时时间没有超时控制的网络请求可能会长时间挂起最终拖慢整个任务。aiohttp可以为连接建立、数据读取和总体请求分别设置超时timeout aiohttp.ClientTimeout( total15, connect5, sock_read10, ) async with aiohttp.ClientSession(timeouttimeout) as session: # 执行请求 pass其中total一次请求允许消耗的总时间connect建立连接的最长等待时间sock_read读取数据时允许等待的时间。在生产环境中不建议完全关闭超时。外部服务出现异常时超时机制是保护任务稳定性的第一道防线。五、加入自动重试机制网络抖动、服务临时不可用和接口限流都可能导致请求失败。对于偶发错误可以采用指数退避策略进行重试。import asyncio import aiohttp async def fetch_with_retry(session, url, max_retries3): for attempt in range(max_retries): try: async with session.get(url) as response: response.raise_for_status() return await response.json() except ( aiohttp.ClientError, asyncio.TimeoutError, ) as error: if attempt max_retries - 1: raise error wait_seconds 2 ** attempt await asyncio.sleep(wait_seconds)三次请求之间的等待时间依次为1 秒、2 秒、4 秒这种策略比固定间隔重试更合理因为它可以给异常服务留出逐步恢复的时间也能降低大量客户端同时重试造成的压力。需要注意并不是所有失败都应该重试。例如参数错误导致的400 Bad Request即使重新发送请求通常也不会成功。比较适合重试的情况包括请求超时连接被临时中断429 Too Many Requests部分5xx服务端错误。六、一个更完整的并发请求模板下面组合并发限制、超时和重试机制形成一个可以直接用于项目的基础模板import asyncio import aiohttp CONCURRENCY 10 MAX_RETRIES 3 async def fetch(session, semaphore, url): async with semaphore: for attempt in range(MAX_RETRIES): try: async with session.get(url) as response: response.raise_for_status() return { url: url, success: True, data: await response.json(), } except ( aiohttp.ClientError, asyncio.TimeoutError, ) as error: if attempt MAX_RETRIES - 1: return { url: url, success: False, error: str(error), } await asyncio.sleep(2 ** attempt) async def main(): urls [ fhttps://example.com/api/items/{item_id} for item_id in range(100) ] semaphore asyncio.Semaphore(CONCURRENCY) timeout aiohttp.ClientTimeout( total15, connect5, sock_read10, ) connector aiohttp.TCPConnector( limitCONCURRENCY ) async with aiohttp.ClientSession( timeouttimeout, connectorconnector, ) as session: tasks [ fetch(session, semaphore, url) for url in urls ] results await asyncio.gather(*tasks) successful [ item for item in results if item[success] ] failed [ item for item in results if not item[success] ] print(f成功{len(successful)}) print(f失败{len(failed)}) if __name__ __main__: asyncio.run(main())这个模板有一个重要特点单个请求失败时不会直接终止全部任务而是将成功和失败结果统一返回方便后续记录日志、补偿重试或人工排查。类似的并发处理方式也适用于批量调用 AI 服务例如文本摘要、内容审核、同言翻译等场景。七、异步并发在 AI 翻译场景中的应用异步并发不仅适合普通的数据接口也很适合文本摘要、内容审核和智能翻译等 AI 应用。以同言翻译为例当系统需要同时处理多段文本或多个翻译任务时可以通过异步请求减少等待时间让不同任务并行执行从而提升整体处理效率。在这类场景中除了控制并发数量还需要重点处理请求超时、接口限流和失败重试等问题。如果某个翻译任务临时失败程序可以单独记录并重新提交而不必让整个批处理流程中断。因此无论是开发翻译工具还是在现有业务中接入多语言能力本文介绍的asyncio、并发限制和自动重试机制都具有一定的参考价值。八、常见问题与避坑建议1. 不要为每个请求创建一个 Session错误示例async def fetch(url): async with aiohttp.ClientSession() as session: async with session.get(url) as response: return await response.json()频繁创建ClientSession会导致连接无法得到有效复用并增加额外开销。更合理的做法是在任务开始时创建一个 Session让所有请求共享。2. asyncio 不适合直接处理 CPU 密集型任务异步编程主要解决网络、磁盘等 I/O 等待问题。如果任务是图片压缩、视频转码或复杂计算单纯改成async并不会明显提速。CPU 密集型任务更适合使用multiprocessingProcessPoolExecutor独立的任务队列与工作进程。3. gather 默认会传播异常当某个任务抛出未处理异常时asyncio.gather()可能直接将异常向外抛出。因此最好在任务内部捕获可预期的网络异常或者使用results await asyncio.gather( *tasks, return_exceptionsTrue, )不过开启return_exceptionsTrue后结果列表中可能同时包含正常数据和异常对象后续处理时需要主动判断类型。4. 注意接口的限流规则并发能力不等于可以忽略对方的接口规范。如果 API 明确限制每秒请求次数还需要增加速率控制而不能只限制同时运行的请求数量。例如限制并发数为 10只能保证同一时刻最多有 10 个请求并不能保证每秒只发送 10 次请求。九、总结将串行请求改造成异步并发往往是批量 API 任务最直接的性能优化方式。但真正可用于生产环境的并发程序不能只写一个asyncio.gather()还需要综合考虑并发数量限制连接复用请求超时自动重试错误隔离服务端限流结果记录与失败补偿。对于 I/O 密集型任务合理使用asyncio和aiohttp可以大幅缩短执行时间。与此同时也要避免盲目提高并发数。稳定、可控、可恢复通常比单纯追求峰值速度更加重要。