Python:Aiohttp 高性能异步的 HTTP 工具库

⚡1、简述

Aiohttp 是一个基于 asyncio 的 Python 异步网络库，支持：

• 🚀 高性能异步 HTTP 客户端和服务端
• 🌍 WebSocket 支持
• 🧩 中间件、路由系统，可快速构建 Web API
• 📦 与 asyncio 完美结合，适合高并发网络应用

一句话总结：

★

aiohttp = Python 的异步版 requests + Flask + WebSocket 支持。

2、核心功能

• ✉️ HTTP 客户端：替代 requests，支持异步请求
• 🖥️ HTTP 服务端：轻量级 Web 框架
• 🔄 WebSocket：双向通信
• 🧰 中间件机制：支持日志、认证、异常处理

安装方式：

pip install aiohttp

如果要启用更快的 HTTP 解析（推荐）：

pip install aiohttp[speedups]

3、实际案例

1️⃣ 异步 HTTP 客户端（GET 请求）

import aiohttp
import asyncio

asyncdeffetch():
asyncwith aiohttp.ClientSession() as session:
asyncwith session.get("https://httpbin.org/get") as resp:
            print("状态码:", resp.status)
            data = await resp.json()
            print("响应数据:", data)

asyncio.run(fetch())

输出示例：

状态码: 200
响应数据: {'args': {}, 'headers': {...}, 'url': 'https://httpbin.org/get'}

2️⃣ 异步 HTTP 客户端（POST 请求）

import aiohttp
import asyncio

asyncdefpost_data():
asyncwith aiohttp.ClientSession() as session:
        payload = {"user": "alice", "action": "login"}
asyncwith session.post("https://httpbin.org/post", json=payload) as resp:
            print(await resp.json())

asyncio.run(post_data())

3️⃣ aiohttp 服务端（最小 Web 服务）

from aiohttp import web

asyncdefhandle(request):
return web.Response(text="Hello, aiohttp!")

app = web.Application()
app.router.add_get("/", handle)

if __name__ == "__main__":
    web.run_app(app, port=8080)

运行后，访问 http://127.0.0.1:8080，即可看到：

Hello, aiohttp!

4️⃣ RESTful API 服务端

from aiohttp import web

asyncdefget_user(request):
    user_id = request.match_info["user_id"]
return web.json_response({"user_id": user_id, "name": "Alice"})

asyncdefcreate_user(request):
    data = await request.json()
return web.json_response({"msg": "User created", "data": data})

app = web.Application()
app.router.add_get("/user/{user_id}", get_user)
app.router.add_post("/user", create_user)

if __name__ == "__main__":
    web.run_app(app, port=8080)

请求示例：

curl http://127.0.0.1:8080/user/42
# {"user_id": "42", "name": "Alice"}

curl -X POST http://127.0.0.1:8080/user -d '{"name":"Bob"}' -H "Content-Type: application/json"
# {"msg": "User created", "data": {"name": "Bob"}}

5️⃣ WebSocket 示例

服务端：

from aiohttp import web

asyncdefwebsocket_handler(request):
    ws = web.WebSocketResponse()
await ws.prepare(request)

asyncfor msg in ws:
if msg.type == web.WSMsgType.TEXT:
await ws.send_str(f"服务端收到: {msg.data}")
elif msg.type == web.WSMsgType.CLOSE:
break

return ws

app = web.Application()
app.router.add_get("/ws", websocket_handler)

if __name__ == "__main__":
    web.run_app(app, port=8080)

客户端：

import aiohttp
import asyncio

asyncdefws_client():
asyncwith aiohttp.ClientSession() as session:
asyncwith session.ws_connect("http://127.0.0.1:8080/ws") as ws:
await ws.send_str("你好 aiohttp!")
            msg = await ws.receive()
            print("收到消息:", msg.data)

asyncio.run(ws_client())

输出：

收到消息: 服务端收到: 你好 aiohttp!

4、应用场景

• 🚀 高性能异步爬虫（替代 requests）
• 🛠️ 轻量级 RESTful API（类似 Flask/FastAPI）
• 📡 WebSocket 实时通信（聊天室、股票推送）
• ☁️ 微服务网关/中间件

4、总结

• aiohttp = 异步 requests + Web 框架 + WebSocket 支持
• 更适合高并发、实时通信的场景
• 如果你熟悉 asyncio，aiohttp 会让你如虎添翼

一句话总结：

★

如果 FastAPI 是异步 Web 框架的明星，那 aiohttp 则是最灵活、最轻量的异步网络工具箱。

📚 参考资料

• 官方文档：https://docs.aiohttp.org/
• GitHub：https://github.com/aio-libs/aiohttp