Aiohttp 基础教程

1. aiohttp 是什么？

aiohttp 是 Python 最主流的异步 HTTP 客户端/服务端库，与 asyncio 搭配使用。

相比 requests：

能力	requests	aiohttp
异步	❌	✅
并发大量请求	❌	✅
文件下载	✅	✅
流式读取	✅	✅
连接池	❌	✅

主要用于：

❗️高并发 HTTP 请求
❗️API 采集/爬虫
❗️异步微服务客户端
❗️下载大文件（流式）

2. 安装

pip install aiohttp

3. 最基本的 GET/POST

GET 请求

import aiohttp
import asyncio

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://httpbin.org/get") as resp:
            print(await resp.json())

asyncio.run(main())

POST 请求（JSON）

async with session.post("https://httpbin.org/post", json={"a": 1}) as resp:
    data = await resp.json()

4. 添加 Headers / Token

headers = {
    "Authorization": "Bearer your-token",
    "User-Agent": "martin/1.0"
}

async with session.get(url, headers=headers) as resp:
    data = await resp.json()

用 FastAPI + JWT，这个写法很常用。

5. Query 参数

params = {"page": 1, "size": 20}

async with session.get(url, params=params) as resp:
    print(await resp.json())

aiohttp 会自动编码。

6. 超时设置

timeout = aiohttp.ClientTimeout(total=10)  # 总超时 10 秒

async with aiohttp.ClientSession(timeout=timeout) as session:
    ...

async with session.get(url, cookies={'sessionid': 'abc123'}) as resp:
    ...

8. 文件下载（流式读取 🔥）

这是 aiohttp 最常用的场景之一：

async def download(url, filepath):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            with open(filepath, "wb") as f:
                async for chunk in resp.content.iter_chunked(1024):
                    f.write(chunk)

asyncio.run(download("https://example.com/file.zip", "file.zip"))

支持大文件，不会爆内存。

9. 上传文件（multipart）

data = aiohttp.FormData()
data.add_field("file", open("a.jpg", "rb"), filename="a.jpg")

async with session.post(url, data=data) as resp:
    print(await resp.json())

10. 使用连接池（推荐用于大量请求）

conn = aiohttp.TCPConnector(limit=100)  # 最大100个连接

async with aiohttp.ClientSession(connector=conn) as session:
    ...

高并发时推荐设置 limit。

11. 并发批量请求（核心功能🔥）

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.text()

async def main():
    urls = [
        "https://httpbin.org/get?a=1",
        "https://httpbin.org/get?a=2",
        "https://httpbin.org/get?a=3",
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, u) for u in urls]
        results = await asyncio.gather(*tasks)
        print(results)

asyncio.run(main())

高并发处理 API / 爬虫，性能非常强。

12. 捕获异常

from aiohttp import ClientError

try:
    async with session.get(url) as resp:
        resp.raise_for_status()
except ClientError as e:
    print(f"请求失败: {e}")

13. aiohttp 服务端（可选）

你可能不常用，FastAPI 更现代一些。

最简服务器：

from aiohttp import web

async def hello(request):
    return web.json_response({"msg": "Hello"})

app = web.Application()
app.add_routes([web.get('/', hello)])

web.run_app(app, port=8000)

14. 最佳实践汇总（适合生产）

场景	推荐写法
高并发 API 调用	统一 ClientSession，使用连接池
文件下载	async for chunk 流式读取
上传	FormData
Token	headers
访问第三方服务	设置超时 + 重试
异步任务池	asyncio.gather

15. 常用模式：会话复用（强烈建议）

class AsyncHTTPClient:
    def __init__(self):
        self.session = aiohttp.ClientSession(
            timeout=aiohttp.ClientTimeout(total=10),
            connector=aiohttp.TCPConnector(limit=100)
        )

    async def get(self, url, **kwargs):
        async with self.session.get(url, **kwargs) as resp:
            return await resp.json()

    async def close(self):
        await self.session.close()

client = AsyncHTTPClient()

这种写法可以直接用在你的 FastAPI 项目里作为单例，性能非常好。