Python 异步编程实战：用 aiohttp 打造高并发 API 客户端

作者 lucy · 2026-04-11

⚠️ 声明：本文相关内容仅供参考，实际效果因场景不同可能差异很大，请结合自身情况判断，谨慎参考。

一、为什么需要异步编程？

在日常开发中，我们经常会遇到这类场景：需要同时调用多个外部 API、批量抓取网页数据、同时处理大量 I/O 操作。如果用传统的同步方式，这些 I/O 等待会白白浪费 CPU 资源。

Python 的 asyncio 模块正是为解决这类问题而生的。配合 aiohttp 库，可以轻松实现单线程高并发，QPS 提升 5～10 倍并不罕见。

二、核心概念：async / await

异步编程的核心是两个关键字：

async def：定义一个协程函数
await：等待另一个协程执行完成（不阻塞事件循环）

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)   # 模拟 I/O 操作
    print("World")

asyncio.run(hello())

三、实战：使用 aiohttp 批量请求

假设我们需要并发请求 100 个 API 端点，同步方式需要逐个等待，总耗时约 100 × 单次耗时。使用 aiohttp 配合信号量控制并发量，既快又稳：

import aiohttp
import asyncio

async def fetch(session, url, sem):
    async with sem:                           # 限制并发数，防止对端被冲垮
        async with session.get(url) as resp:
            return await resp.json()

async def batch_fetch(urls):
    sem = asyncio.Semaphore(20)              # 最多同时 20 个请求
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url, sem) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

urls = [f"https://api.example.com/item/{i}" for i in range(100)]
results = asyncio.run(batch_fetch(urls))

以上代码在几秒内即可完成 100 个网络请求，比同步方式快了一个数量级。

四、生产级技巧

超时控制：使用 aiohttp.ClientTimeout(total=10) 防止单点请求卡死整个任务
重试机制：配合 tenacity 库对失败请求自动重试，提高整体成功率
连接池复用：一个 ClientSession 复用 TCP 连接，减少握手开销
异常处理：gather(*tasks, return_exceptions=True) 保证个别失败不影响全局

五、总结

异步编程是 Python 后端工程师不可忽视的技能，尤其在 I/O 密集型场景下收益明显。上手门槛不高，建议从 aiohttp 批量请求练起，逐步迁移到异步数据库（asyncpg）、异步 Redis（aioredis）等场景，构建真正高性能的异步服务。

📡 来源

📡 来源：笔者实战经验整理，2026-04-16