写爬虫或者做接口调用的时候，很多人第一反应就是 import requests。不可否认，Requests 的 API 设计得非常优雅，极其好用。但是，当你需要同时抓取几百个网页，或者高频调用大量 API 时，单纯依靠同步阻塞的 Requests 会让你的程序慢得让人抓狂。
今天我们要聊的主角是 aiohttp。如果说 Requests 是一个人在窗口排队办事，那么 aiohttp 就是你学会了影分身之术，同时去几百个窗口办事。它是一个基于 asyncio 的异步 HTTP 网络模块，不仅支持客户端请求，还能用来搭建服务端。
为了直观感受两者的区别，我们先来看一张流程图。
同步 vs 异步工作流程图
【传统同步模式 requests】[发起请求 A] ──> 原地死等 ──> [处理响应 A] └──> [发起请求 B] ──> 原地死等 ──> [处理响应 B](评价：CPU 和网络带宽都在白白浪费时间等待)【异步高并发模式 aiohttp】[发起请求 A] ──> 挂起等待响应，立刻去做别的事 ──> [收到 A 响应] ──> [处理响应 A][发起请求 B] ──> 挂起等待响应，立刻去做别的事 ──> [收到 B 响应] ──> [处理响应 B][发起请求 C] ──> 挂起等待响应，立刻去做别的事 ──> [收到 C 响应] ──> [处理响应 C](评价：请求几乎同时发出，完美榨干网络 I/O 性能，谁先返回就先处理谁)
搞清楚了原理，我们直接进入代码实战。
基础入门：发起一个异步请求
使用 aiohttp 的核心思想是建立一个全局的 ClientSession（会话），然后用这个会话去发起各种请求。注意，所有涉及网络 I/O 的地方，都要加上 await 关键字。
import aiohttpimport asyncioasync def fetch_single_page(): # 推荐使用 async with 管理上下文，确保用完后自动关闭连接 async with aiohttp.ClientSession() as session: # 发起 GET 请求 url = 'https://httpbin.org/get' async with session.get(url) as response: # 打印状态码 print(f"Status Code: {response.status}") # 读取响应内容，这里也是一个异步操作 text = await response.text() print(f"Response Body: {text[:100]}...")# 运行异步函数if __name__ == '__main__': asyncio.run(fetch_single_page())
这段代码看起来比 requests 稍微繁琐一点，但这是为了后续的高并发打下基础。记住这个套路：外层建 Session，内层发 Request，读取内容加 await。
火力全开：高并发批量请求
这是 aiohttp 真正发光发热的地方。假设我们现在有 10 个 API 接口需要请求，用 requests 需要一个个等，而用 aiohttp，我们可以用 asyncio.gather 把所有任务打包，一次性发射出去。
import aiohttpimport asyncioimport timeasync def fetch_data(session, url, task_id): """单独的请求任务""" print(f"任务 {task_id} 开始请求...") async with session.get(url) as response: await response.read() # 模拟读取数据 print(f"任务 {task_id} 完成！状态码: {response.status}")async def main(): url = 'https://httpbin.org/delay/1' # 这是一个会延迟1秒才响应的测试接口 # 开启大同级会话 async with aiohttp.ClientSession() as session: tasks = [] # 创建 10 个并发任务 for i in range(1, 11): task = fetch_data(session, url, i) tasks.append(task) # gather 会并发执行所有任务，并等待它们全部完成 await asyncio.gather(*tasks)if __name__ == '__main__': start_time = time.time() asyncio.run(main()) print(f"总耗时: {time.time() - start_time:.2f} 秒") # 结果剧透：如果用 requests，耗时一定是 10 秒以上。 # 用 aiohttp，总耗时通常在 1.5 秒左右！
进阶实战：超时控制与请求头设置
在真实的生产环境中，网络情况极其复杂。有些接口可能会卡死，有些网站会校验 User-Agent。aiohttp 提供了非常精细的控制选项。
我们可以通过 aiohttp.ClientTimeout 来设置超时，同时在请求时传入 Headers 或参数。
import aiohttpimport asyncioasync def fetch_with_params(): url = 'https://httpbin.org/get' # 自定义请求头 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)', 'Authorization': 'Bearer your_token_here' } # URL 参数 params = { 'search': 'python aiohttp', 'page': '1' } # 设置总超时时间为 5 秒，连接超时为 2 秒 timeout = aiohttp.ClientTimeout(total=5, connect=2) try: # 在发起请求时传入 headers, params 和 timeout async with aiohttp.ClientSession(timeout=timeout, headers=headers) as session: async with session.get(url, params=params) as response: # 如果状态码不是 2xx，直接抛出异常，类似 requests.raise_for_status() response.raise_for_status() # 直接将响应解析为 JSON 字典 json_data = await response.json() print("解析出的 JSON 数据：") print(json_data['args']) # 打印我们传过去的参数 except asyncio.TimeoutError: print("请求超时了，请检查网络或代理！") except aiohttp.ClientError as e: print(f"网络请求发生错误: {e}")if __name__ == '__main__': asyncio.run(fetch_with_params())
总结与避坑指南
当你习惯了异步编程的思维逻辑后，aiohttp 会成为你兵器库里最锋利的刀。不过在实际使用中，有几个坑需要特别注意：
不要在每次请求时都新建 ClientSession。 ClientSession 内部维护了连接池，频繁创建和销毁会极其消耗资源。正确的做法是全局只建一个，然后在不同的方法之间传递它。
控制并发量。 虽然 aiohttp 可以轻松发起成千上万个请求，但如果目标服务器承受不住，你就会被封 IP。建议配合 asyncio.Semaphore (信号量) 来限制同时发出的并发数量，做到温柔采血。
注意阻塞操作。 在 async def 定义的异步函数里，千万不能使用 time.sleep() 这种同步阻塞代码，它会卡死整个事件循环，必须换成 await asyncio.sleep()。
拥抱异步，让你的 Python 脚本飞起来吧！如果在实际操作中遇到什么奇怪的报错，欢迎在评论区留言交流。
10个月宝宝每天需要喝多少奶粉？