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

📅 2026/7/18 20:17:25 👁️ 阅读次数 📝 编程学习
Python 并发请求实战:用 asyncio 把批量 API 调用提速 10 倍

在数据采集、AI 模型调用、文件处理等项目中,我们经常需要批量请求 API。最直接的写法是使用requests逐个发送请求,但当任务数量达到几百甚至几千时,串行执行会浪费大量时间。

这篇文章将从一个简单案例出发,介绍如何使用 Python 的asyncioaiohttp实现并发请求,并补充超时控制、限流、重试等实际项目中不可缺少的机制。

一、为什么串行请求速度慢?

假设我们需要请求 100 个接口,每个接口平均响应时间为 500 毫秒。使用串行方式时,总耗时大约为:

100 × 0.5 秒 = 50 秒

传统写法如下:

import requests def fetch(url): response = requests.get(url, timeout=10) response.raise_for_status() return response.json() urls = [ f"https://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 = [ f"https://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())

这里有三个关键点:

  1. 使用async def定义异步函数。

  2. 使用await等待网络请求完成。

  3. 使用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 个请求。其他任务会进入等待状态,直到有可用的并发名额。

并发数没有适用于所有项目的固定答案,可以从 5~20 开始测试,再根据接口限流规则、服务器性能和实际响应时间进行调整。

四、为请求设置超时时间

没有超时控制的网络请求可能会长时间挂起,最终拖慢整个任务。

aiohttp可以为连接建立、数据读取和总体请求分别设置超时:

timeout = aiohttp.ClientTimeout( total=15, connect=5, sock_read=10, ) async with aiohttp.ClientSession(timeout=timeout) as session: # 执行请求 pass

其中:

  • total:一次请求允许消耗的总时间;

  • connect:建立连接的最长等待时间;

  • sock_read:读取数据时允许等待的时间。

在生产环境中,不建议完全关闭超时。外部服务出现异常时,超时机制是保护任务稳定性的第一道防线。

五、加入自动重试机制

网络抖动、服务临时不可用和接口限流都可能导致请求失败。对于偶发错误,可以采用指数退避策略进行重试。

import asyncio import aiohttp async def fetch_with_retry(session, url, max_retries=3): 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 = [ f"https://example.com/api/items/{item_id}" for item_id in range(100) ] semaphore = asyncio.Semaphore(CONCURRENCY) timeout = aiohttp.ClientTimeout( total=15, connect=5, sock_read=10, ) connector = aiohttp.TCPConnector( limit=CONCURRENCY ) async with aiohttp.ClientSession( timeout=timeout, connector=connector, ) 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 密集型任务更适合使用:

  • multiprocessing

  • ProcessPoolExecutor

  • 独立的任务队列与工作进程。

3. gather 默认会传播异常

当某个任务抛出未处理异常时,asyncio.gather()可能直接将异常向外抛出。因此,最好在任务内部捕获可预期的网络异常,或者使用:

results = await asyncio.gather( *tasks, return_exceptions=True, )

不过,开启return_exceptions=True后,结果列表中可能同时包含正常数据和异常对象,后续处理时需要主动判断类型。

4. 注意接口的限流规则

并发能力不等于可以忽略对方的接口规范。如果 API 明确限制每秒请求次数,还需要增加速率控制,而不能只限制同时运行的请求数量。

例如,限制并发数为 10,只能保证同一时刻最多有 10 个请求,并不能保证每秒只发送 10 次请求。

九、总结

将串行请求改造成异步并发,往往是批量 API 任务最直接的性能优化方式。但真正可用于生产环境的并发程序,不能只写一个asyncio.gather(),还需要综合考虑:

  • 并发数量限制;

  • 连接复用;

  • 请求超时;

  • 自动重试;

  • 错误隔离;

  • 服务端限流;

  • 结果记录与失败补偿。

对于 I/O 密集型任务,合理使用asyncioaiohttp可以大幅缩短执行时间。与此同时,也要避免盲目提高并发数。稳定、可控、可恢复,通常比单纯追求峰值速度更加重要。