Python aiohttp 全面指南：異步HTTP客戶端/服務器框架

邊寫代碼零食不停口盼盼麥香雞味塊、卡樂比（Calbee）薯條三兄弟獨立小包、好時kisses多口味巧克力糖、老金磨方【黑金系列】黑芝麻丸

邊寫代碼邊貼面膜事業美麗兩不誤 DR. YS 野森博士+【AOUFSE/澳芙雪特證】377專研美白淡斑面膜組合優惠劵

別光顧寫代碼更要多喝茶水，提神有營養六安瓜片茶葉茶香二級200g 2025年新茶雨前盒裝自己喝

讓AI成為我們的得力助手：《用Cursor玩轉AI輔助編程——不寫代碼也能做軟件開發》

Python 圖書推薦

書名	出版社	推薦
Python編程從入門到實踐第3版（圖靈出品）	人民郵電出版社	★★★★★
Python數據科學手冊（第2版）（圖靈出品）	人民郵電出版社	★★★★★
圖形引擎開發入門：基于Python語言	電子工業出版社	★★★★★
科研論文配圖繪制指南基于Python（異步圖書出品）	人民郵電出版社	★★★★★
Effective Python：編寫好Python的90個有效方法（第2版英文版）	人民郵電出版社	★★★★★
Python人工智能與機器學習（套裝全5冊)	清華大學出版社	★★★★★

JAVA 圖書推薦

書名	出版社	推薦
Java核心技術第12版：卷Ⅰ+卷Ⅱ	機械工業出版社	★★★★★
Java核心技術第11版套裝共2冊	機械工業出版社	★★★★★
Java語言程序設計基礎篇+進階篇原書第12版套裝共2冊	機械工業出版社	★★★★★
Java 11官方參考手冊（第11版）	清華大學出版社	★★★★★
Offer來了：Java面試核心知識點精講（第2版）(博文視點出品)	電子工業出版社	★★★★★

什么是 aiohttp？

aiohttp 是一個基于 Python asyncio 的異步 HTTP 客戶端/服務器框架，專為高性能網絡編程設計。它提供了：

異步 HTTP 客戶端（類似異步版 requests）
異步 HTTP 服務器（類似異步版 Flask/Django）
完整的 WebSocket 支持
高效的連接池管理

核心優勢

特性	描述
異步非阻塞	單線程處理數千并發連接
高性能	遠超同步框架（如 requests）的吞吐量
輕量級	簡潔的API，無復雜依賴
全面協議支持	HTTP/1.1, HTTP/2(客戶端), WebSocket
生態完善	良好文檔和活躍社區

基礎用法 - HTTP客戶端

安裝

pip install aiohttp

基本GET請求

import aiohttp
import asyncioasync def main():async with aiohttp.ClientSession() as session:async with session.get('https://api.example.com/data') as response:print("狀態碼:", response.status)print("響應內容:", await response.text())asyncio.run(main())

POST請求示例

async def post_example():async with aiohttp.ClientSession() as session:# 表單數據async with session.post('https://httpbin.org/post', data={'key': 'value'}) as response:print(await response.json())# JSON數據async with session.post('https://api.example.com/users',json={'name': 'Alice', 'age': 30}) as response:print(await response.json())

高級用法

并發請求

async def fetch(url):async with aiohttp.ClientSession() as session:async with session.get(url) as response:return await response.text()async def concurrent_requests():urls = ['https://api.example.com/item/1','https://api.example.com/item/2','https://api.example.com/item/3']tasks = [fetch(url) for url in urls]results = await asyncio.gather(*tasks)for url, content in zip(urls, results):print(f"{url}: {content[:50]}...")asyncio.run(concurrent_requests())

超時控制

async def timeout_example():timeout = aiohttp.ClientTimeout(total=5)  # 5秒總超時async with aiohttp.ClientSession(timeout=timeout) as session:try:async with session.get('https://slow-api.example.com') as response:return await response.text()except asyncio.TimeoutError:print("請求超時!")

流式處理大響應

async def stream_response():async with aiohttp.ClientSession() as session:async with session.get('https://large-file.example.com') as response:with open('large_file.txt', 'wb') as f:async for chunk in response.content.iter_chunked(1024):f.write(chunk)print(f"已接收 {len(chunk)} 字節")

服務器端開發

基本HTTP服務器

from aiohttp import webasync def handle(request):name = request.match_info.get('name', "World")return web.Response(text=f"Hello, {name}!")app = web.Application()
app.add_routes([web.get('/', handle),web.get('/{name}', handle)
])if __name__ == '__main__':web.run_app(app, port=8080)

REST API示例

async def get_users(request):users = [{'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}]return web.json_response(users)async def create_user(request):data = await request.json()# 實際應用中這里會保存到數據庫return web.json_response({'id': 3, **data}, status=201)app = web.Application()
app.add_routes([web.get('/api/users', get_users),web.post('/api/users', create_user)
])

WebSocket服務器

async def websocket_handler(request):ws = web.WebSocketResponse()await ws.prepare(request)async for msg in ws:if msg.type == aiohttp.WSMsgType.TEXT:if msg.data == 'close':await ws.close()else:await ws.send_str(f"ECHO: {msg.data}")elif msg.type == aiohttp.WSMsgType.ERROR:print('WebSocket連接異常關閉')return wsapp.add_routes([web.get('/ws', websocket_handler)])

進階擴展

中間件示例

async def auth_middleware(app, handler):async def middleware(request):# 驗證API密鑰if request.headers.get('X-API-Key') != 'SECRET_KEY':return web.json_response({'error': 'Unauthorized'}, status=401)return await handler(request)return middlewareapp = web.Application(middlewares=[auth_middleware])

HTTP/2客戶端支持

async def http2_request():conn = aiohttp.TCPConnector(force_close=True, enable_cleanup_closed=True)async with aiohttp.ClientSession(connector=conn) as session:async with session.get('https://http2.akamai.com/',headers={'accept': 'text/html'}) as response:print("HTTP版本:", response.version)print("內容:", await response.text()[:200])

性能優化配置

# 自定義連接器配置
connector = aiohttp.TCPConnector(limit=100,  # 最大并發連接數limit_per_host=20,  # 單主機最大連接數ssl=False,  # 禁用SSL驗證（僅用于測試）force_close=True  # 避免連接延遲關閉
)# 自定義會話配置
session = aiohttp.ClientSession(connector=connector,timeout=aiohttp.ClientTimeout(total=30),headers={'User-Agent': 'MyApp/1.0'},cookie_jar=aiohttp.CookieJar(unsafe=True)
)

最佳實踐

重用ClientSession：避免為每個請求創建新會話
使用連接池：合理配置TCPConnector參數
超時設置：總是配置合理的超時時間
資源清理：使用async with確保資源釋放

錯誤處理：捕獲并處理常見網絡異常

try:async with session.get(url) as response:response.raise_for_status()return await response.json()
except aiohttp.ClientError as e:print(f"請求錯誤: {e}")

完整示例

import aiohttp
import asyncio
from aiohttp import web# 客戶端示例
async def fetch_data():async with aiohttp.ClientSession() as session:# 并發請求多個APIurls = ['https://jsonplaceholder.typicode.com/posts/1','https://jsonplaceholder.typicode.com/comments/1','https://jsonplaceholder.typicode.com/albums/1']tasks = []for url in urls:tasks.append(session.get(url))responses = await asyncio.gather(*tasks)results = []for response in responses:results.append(await response.json())return results# 服務器示例
async def handle_index(request):return web.Response(text="Welcome to aiohttp server!")async def handle_api(request):data = await fetch_data()return web.json_response(data)# 創建應用
app = web.Application()
app.add_routes([web.get('/', handle_index),web.get('/api', handle_api)
])# 啟動服務器
async def start_server():runner = web.AppRunner(app)await runner.setup()site = web.TCPSite(runner, 'localhost', 8080)await site.start()print("Server running at http://localhost:8080")# 保持運行while True:await asyncio.sleep(3600)  # 每小時喚醒一次if __name__ == '__main__':asyncio.run(start_server())