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, 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 密集型任务，合理使用asyncio和aiohttp可以大幅缩短执行时间。与此同时，也要避免盲目提高并发数。稳定、可控、可恢复，通常比单纯追求峰值速度更加重要。