大家好，我是jobleap.cn的小九。
aiohttp 是 Python 生态中最主流的异步 HTTP 客户端/服务器框架，基于 asyncio 实现，支持异步 HTTP 请求、WebSocket 通信、HTTP 服务器开发等核心能力，广泛应用于高并发爬虫、异步 API 服务开发等场景。本教程将从环境准备到核心 API 实战，再到综合案例，全面串联 aiohttp 的常用功能，确保你能一站式掌握其核心用法。
一、环境准备
1. 版本要求
- Python 版本：≥3.7（推荐 3.8+，asyncio 特性更稳定）
- aiohttp 版本：推荐最新稳定版（当前主流为 3.9+）
2. 安装
pipinstallaiohttp# 核心库pipinstallaiohttp[speedups]# 安装加速依赖（可选）二、aiohttp 客户端核心 API 实战
客户端是 aiohttp 最常用的场景，用于异步发送 HTTP 请求（GET/POST/PUT/DELETE 等），核心载体是aiohttp.ClientSession（建议复用 Session 以节省资源）。
1. 基础异步 GET 请求
核心 API：ClientSession.get()+async with上下文管理（自动释放连接）。
importasyncioimportaiohttpasyncdefbasic_get_request():# 1. 创建 ClientSession（复用连接，建议全局/单例）asyncwithaiohttp.ClientSession()assession:# 2. 发送 GET 请求（async with 自动处理响应关闭）asyncwithsession.get("https://httpbin.org/get")asresponse:# 3. 响应基础信息print("状态码:",response.status)# 200print("响应头:",dict(response.headers))# 4. 解析响应内容（text()/json()/read()）text_content=awaitresponse.text()# 字符串形式json_content=awaitresponse.json()# JSON 解析（自动转字典）# bytes_content = await response.read() # 字节形式print("JSON 响应内容:",json_content)# 运行异步函数if__name__=="__main__":asyncio.run(basic_get_request())2. 带参数的 GET 请求
核心 API：params参数（自动拼接 URL 查询字符串）。
asyncdefget_with_params():asyncwithaiohttp.ClientSession()assession:# 构造查询参数（字典/列表/元组）params={"name":"aiohttp","version":"3.9","page":1}# 发送带参数的 GET 请求（最终 URL：https://httpbin.org/get?name=aiohttp&version=3.9&page=1）asyncwithsession.get(url="https://httpbin.org/get",params=params)asresponse:data=awaitresponse.json()print("请求参数验证:",data["args"])# 输出传入的 paramsasyncio.run(get_with_params())3. 异步 POST 请求
核心 API：ClientSession.post()，支持两种常见数据格式：表单（form-data）、JSON。
3.1 发送表单数据（form-data）
asyncdefpost_form_data():asyncwithaiohttp.ClientSession()assession:# 构造表单数据form_data=aiohttp.FormData()form_data.add_field("username","test_user")form_data.add_field("password","123456")# 可选：上传文件（form-data 场景）# form_data.add_field("file", open("test.txt", "rb"), filename="test.txt")# 发送 POST 请求（data 参数传 FormData）asyncwithsession.post(url="https://httpbin.org/post",data=form_data)asresponse:data=awaitresponse.json()print("表单数据响应:",data["form"])# 输出 {"username":"test_user","password":"123456"}asyncio.run(post_form_data())3.2 发送 JSON 数据
asyncdefpost_json_data():asyncwithaiohttp.ClientSession()assession:# 构造 JSON 数据（字典）json_data={"title":"aiohttp 教程","content":"异步 HTTP 客户端","tags":["python","asyncio","aiohttp"]}# 发送 JSON 请求（json 参数自动序列化并设置 Content-Type: application/json）asyncwithsession.post(url="https://httpbin.org/post",json=json_data)asresponse:data=awaitresponse.json()print("JSON 数据响应:",data["json"])# 输出传入的 json_dataasyncio.run(post_json_data())4. 请求头与 Cookie 配置
核心 API：headers参数（自定义请求头）、cookies参数（设置 Cookie）。
asyncdefcustom_headers_cookies():# 1. 自定义请求头headers={"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) aiohttp/3.9","Referer":"https://example.com","Authorization":"Bearer your_token_here"# 认证 Token}# 2. 自定义 Cookiecookies={"session_id":"123456789","user_lang":"zh-CN"}asyncwithaiohttp.ClientSession(headers=headers,# 全局请求头（Session 级别，所有请求生效）cookies=cookies# 全局 Cookie（Session 级别）)assession:# 局部请求头（覆盖全局）asyncwithsession.get("https://httpbin.org/headers",headers={"X-Custom-Header":"aiohttp-test"})asresponse:data=awaitresponse.json()print("请求头验证:",data["headers"])print("Cookie 验证:",awaitsession.cookie_jar.filter_cookies("https://httpbin.org"))5. 超时控制
核心 API：aiohttp.ClientTimeout（全局/局部超时），避免请求无限阻塞。
asyncdeftimeout_control():# 1. 定义超时配置（总超时 5 秒，连接超时 2 秒，读取超时 3 秒）timeout=aiohttp.ClientTimeout(total=5,# 总超时（推荐设置）connect=2,# 连接超时sock_read=3# 读取响应超时)asyncwithaiohttp.ClientSession(timeout=timeout)assession:try:# 测试超时（访问一个慢接口）asyncwithsession.get("https://httpbin.org/delay/10")asresponse:print(awaitresponse.text())exceptasyncio.TimeoutError:print("请求超时！")exceptaiohttp.ClientErrorase:print("客户端错误:",e)asyncio.run(timeout_control())6. 并发请求（批量发送）
核心 API：asyncio.gather()（并发执行多个异步请求），结合ClientSession实现高并发。
asyncdeffetch_url(session,url):"""单个 URL 请求函数"""try:asyncwithsession.get(url)asresponse:return{"url":url,"status":response.status,"content":awaitresponse.text()[:50]}exceptExceptionase:return{"url":url,"error":str(e)}asyncdefbatch_requests():# 待请求的 URL 列表urls=["https://httpbin.org/get","https://httpbin.org/delay/1",# 延迟 1 秒"https://httpbin.org/status/404",# 404 状态码"https://invalid-url.example.com"# 无效 URL]asyncwithaiohttp.ClientSession()assession:# 构造任务列表tasks=[fetch_url(session,url)forurlinurls]# 并发执行所有任务（gather 等待所有任务完成）results=awaitasyncio.gather(*tasks)# 打印结果forresinresults:print("并发请求结果:",res)asyncio.run(batch_requests())7. 异常处理
aiohttp 客户端常见异常：
aiohttp.ClientError：所有客户端异常的基类aiohttp.ClientConnectionError：连接错误（如网络不通）aiohttp.ClientResponseError：响应错误（如 4xx/5xx）asyncio.TimeoutError：超时错误
asyncdefexception_handling():asyncwithaiohttp.ClientSession()assession:try:asyncwithsession.get("https://httpbin.org/status/404")asresponse:# 主动检查状态码（非 2xx/3xx 抛出异常）response.raise_for_status()print(awaitresponse.text())exceptaiohttp.ClientResponseErrorase:print(f"响应错误: 状态码{e.status}, 消息{e.message}")exceptaiohttp.ClientConnectionErrorase:print(f"连接错误:{e}")exceptasyncio.TimeoutError:print("请求超时")exceptaiohttp.ClientErrorase:print(f"客户端通用错误:{e}")asyncio.run(exception_handling())三、aiohttp 服务器核心 API 实战
aiohttp 也支持搭建异步 HTTP 服务器，核心载体是aiohttp.web.Application，适用于轻量级 API 服务开发。
1. 基础 HTTP 服务器搭建
核心 API：web.Application()+web.run_app()+ 路由装饰器@routes.get/post。
fromaiohttpimportweb# 1. 定义请求处理函数（接收 request 参数，返回 Response）asyncdefhello_handler(request):# 返回文本响应（默认状态码 200）returnweb.Response(text="Hello, aiohttp Server!")# 2. 创建 Application 实例（核心容器）app=web.Application()# 3. 注册路由（URL 路径 → 处理函数）app.add_routes([web.get("/hello",hello_handler),# GET 请求 → /hello 路径])# 4. 运行服务器（默认端口 8080，host 0.0.0.0）if__name__=="__main__":web.run_app(app,host="0.0.0.0",port=8080)运行后访问http://localhost:8080/hello，将返回Hello, aiohttp Server!。
2. 路由管理（静态/动态路由）
2.1 动态路由参数
核心 API：路由路径中使用{param}定义参数，通过request.match_info获取。
asyncdefuser_handler(request):# 获取动态路由参数user_id=request.match_info.get("user_id")username=request.match_info.get("username","guest")# 带默认值returnweb.Response(text=f"User ID:{user_id}, Username:{username}")# 注册动态路由app.add_routes([web.get("/user/{user_id}/{username}",user_handler),# 多参数web.get("/user/{user_id}",user_handler),# 单参数])访问http://localhost:8080/user/123/aiohttp，返回User ID: 123, Username: aiohttp。
2.2 路由列表批量注册
# 定义多个处理函数asyncdefhome_handler(request):returnweb.Response(text="Home Page")asyncdefabout_handler(request):returnweb.Response(text="About aiohttp Server")# 批量注册路由routes=web.RouteTableDef()# 路由表routes.add_get("/",home_handler)routes.add_get("/about",about_handler)app.add_routes(routes)# 注册到应用3. 请求参数解析
服务器端解析客户端传入的参数（GET 查询参数、POST JSON/表单数据）。
3.1 解析 GET 查询参数
asyncdefget_params_handler(request):# 获取 GET 查询参数（如 /query?name=aiohttp&page=1）name=request.query.get("name","unknown")page=int(request.query.get("page",1))returnweb.Response(text=f"Name:{name}, Page:{page}")app.add_routes([web.get("/query",get_params_handler)])3.2 解析 POST JSON 数据
asyncdefpost_json_handler(request):try:# 解析 JSON 数据（自动转字典）data=awaitrequest.json()title=data.get("title")content=data.get("content")returnweb.json_response({"status":"success","data":{"title":title,"content":content}})exceptExceptionase:# 返回 JSON 错误响应（自定义状态码）returnweb.json_response({"error":str(e)},status=400)app.add_routes([web.post("/json",post_json_handler)])3.3 解析 POST 表单数据
asyncdefpost_form_handler(request):# 解析表单数据data=awaitrequest.post()username=data.get("username")password=data.get("password")returnweb.Response(text=f"Username:{username}, Password:{password}")app.add_routes([web.post("/form",post_form_handler)])4. 中间件（Middleware）
中间件用于拦截请求/响应，实现日志、认证、异常处理等通用逻辑，核心 API：web.Application(middlewares=[])。
# 定义日志中间件asyncdeflog_middleware(app,handler):asyncdeflog_handler(request):# 请求前日志print(f"Request:{request.method}{request.path}")# 执行原处理函数response=awaithandler(request)# 响应后日志print(f"Response:{response.status}")returnresponsereturnlog_handler# 定义异常处理中间件asyncdeferror_middleware(app,handler):asyncdeferror_handler(request):try:returnawaithandler(request)exceptExceptionase:returnweb.json_response({"error":str(e)},status=500)returnerror_handler# 创建应用并注册中间件（执行顺序：先注册的先执行）app=web.Application(middlewares=[log_middleware,error_middleware])5. 静态文件服务
核心 API：web.static()，用于提供静态文件（HTML/CSS/JS/图片等）。
# 注册静态文件路由（/static 路径映射到本地 static 目录）app.add_routes([web.static("/static","./static")# 本地目录 ./static → http://localhost:8080/static/xxx])将静态文件（如index.html）放入./static目录，访问http://localhost:8080/static/index.html即可访问。
四、综合实战：异步爬虫 + 简易 API 服务器
本案例串联 aiohttp 客户端（爬虫）和服务器（API）：
- 服务器端：提供两个接口（/api/data 返回模拟数据，/api/save 接收爬虫数据并存储）；
- 客户端：异步爬取第三方数据，调用服务器 /api/save 接口提交数据，再调用 /api/data 接口验证存储结果。
1. 服务器端实现（server.py）
fromaiohttpimportweb# 模拟存储数据的列表storage=[]# 1. 返回模拟数据接口asyncdefget_data_handler(request):returnweb.json_response({"data":storage,"count":len(storage)})# 2. 接收爬虫数据并存储asyncdefsave_data_handler(request):try:data=awaitrequest.json()storage.append(data)returnweb.json_response({"status":"success","message":"数据已保存"})exceptExceptionase:returnweb.json_response({"status":"error","message":str(e)},status=400)# 3. 创建应用并注册路由app=web.Application()app.add_routes([web.get("/api/data",get_data_handler),web.post("/api/save",save_data_handler)])# 4. 运行服务器if__name__=="__main__":web.run_app(app,host="0.0.0.0",port=8080)2. 客户端实现（client.py）
importasyncioimportaiohttpasyncdefcrawl_and_save():# 1. 异步爬取第三方测试数据asyncwithaiohttp.ClientSession()assession:# 爬取 httpbin 的测试数据asyncwithsession.get("https://httpbin.org/get?source=aiohttp_crawler")asresponse:crawl_data=awaitresponse.json()print("爬取的数据:",crawl_data)# 2. 调用服务器 /api/save 接口提交数据asyncwithsession.post("http://localhost:8080/api/save",json=crawl_data)asresponse:save_result=awaitresponse.json()print("保存结果:",save_result)# 3. 调用服务器 /api/data 接口验证数据asyncwithsession.get("http://localhost:8080/api/data")asresponse:verify_data=awaitresponse.json()print("验证存储数据:",verify_data)if__name__=="__main__":asyncio.run(crawl_and_save())3. 运行与验证
- 先启动服务器：
python server.py； - 再运行客户端：
python client.py； - 客户端输出会显示：爬取数据 → 保存成功 → 验证存储的结果；
- 也可直接访问
http://localhost:8080/api/data查看存储的数据。
五、总结与进阶方向
1. 核心 API 总结
| 模块 | 核心 API/类 | 核心用途 |
|---|---|---|
| 客户端 | ClientSession | 异步 HTTP 请求会话（复用连接） |
| get()/post()/put()/delete() | 发送对应方法的 HTTP 请求 | |
| ClientTimeout | 超时控制 | |
| response.text()/json() | 响应数据解析 | |
| 服务器 | web.Application | 服务器核心容器 |
| web.RouteTableDef | 路由表管理 | |
| web.Response/json_response | 构建响应 | |
| request.query/match_info | 解析请求参数 | |
| 通用 | asyncio.gather() | 并发执行异步任务 |
2. 进阶方向
- WebSocket 通信：aiohttp 支持异步 WebSocket 客户端/服务器（
ClientSession.ws_connect()/web.WebSocketResponse）； - 连接池配置：自定义 ClientSession 的连接池大小、超时、代理等；
- HTTPS 证书处理：忽略证书验证、自定义 CA 证书；
- 性能优化：结合
aiohttp[speedups]、限制并发数（避免请求过多）； - 生产级服务器：配合 gunicorn + aiohttp 运行服务器，而非
web.run_app()。
通过本教程，你已掌握 aiohttp 客户端和服务器的核心常用 API，且通过综合案例实现了两者的串联。实际开发中，可根据场景（爬虫/API 服务）灵活组合这些 API，充分发挥异步编程的高并发优势。
