前言
上一篇我们学习了 asyncio 的核心概念，并了解了 aiohttp 的基本用法。本文深入讲解 aiohttp——Python 生态中最成熟的异步 HTTP 框架。我们将用它构建一个功能完整的异步 REST API，包括路由、中间件、WebSocket 支持、静态文件服务，并与同步框架进行性能对比。
aiohttp 核心架构
安装与项目结构
pip install aiohttp
# 可选依赖
pip install aiohttp[speedups] # 启用 yarl 和 async-timeout 加速
project/
├── app.py # 应用入口
├── config.py # 配置管理
├── middlewares.py # 中间件
├── routes/
│ └── api.py # API 路由
├── static/ # 静态文件
└── templates/ # Jinja2 模板
Hello World —— 最简异步服务器
from aiohttp import web
async def hello(request):
return web.Response(text='Hello, aiohttp!')
app = web.Application()
app.router.add_get('/', hello)
if __name__ == '__main__':
web.run_app(app, host='localhost', port=8080)
Request / Response 详解
from aiohttp import web
import time
async def handle_request(request):
# URL 参数
user_id = request.match_info.get('id', '?')
# Query String ?name=John&age=25
name = request.query.get('name', 'World')
# Headers
auth_token = request.headers.get('Authorization', '')
content_type = request.content_type
# Body 读取
data = await request.json() # JSON body
text = await request.text() # Text body
binary = await request.read() # Binary body
# File upload
reader = await request.multipart()
async for field in reader:
if field.name == 'avatar':
content = await field.read(decode=True)
# 响应
return web.Response(text=f'Hello, {name}!')
return web.json_response({'status': 'ok', 'data': {'user_id': user_id}})
return web.HTTPFound('/new-page')
resp = web.Response(status=201, body=b'Created')
return resp
路由系统
基本路由
from aiohttp import web
def setup_routes(app):
app.router.add_get('/', index_handler)
app.router.add_get('/users', list_users)
app.router.add_get('/users/{id}', get_user)
app.router.add_post('/users', create_user)
app.router.add_put('/users/{id}', update_user)
app.router.add_delete('/users/{id}', delete_user)
app.router.add_route('*', '/catch-all', catch_all_handler)
async def get_user(request):
user_id = request.match_info['id']
return web.json_response({'id': user_id, 'name': '张三'})
async def create_user(request):
data = await request.json()
return web.json_response({'created': data}, status=201)
async def catch_all_handler(request):
return web.Response(text=f'{request.method} {request.path}')
带参数的路由
def setup_param_routes(app):
# 字符串参数
app.router.add_get('/users/{username}', get_user_by_name)
# 整数参数
app.router.add_get('/posts/{post_id:int}', get_post)
# 路径通配符（捕获所有剩余部分）
app.router.add_get('/files/{path:path}', serve_files)
# 自定义正则参数
app.router.add_get('/item/{sku:[A-Z]{3}-\\d{4}}', get_item)
async def serve_files(request):
path = request.match_info['path']
return web.FileResponse(f'/tmp/uploads/{path}')
Blueprint —— 模块化路由
class RouterGroup:
"""类似 Flask Blueprint 的路由分组"""
def __init__(self, url_prefix='/api'):
self.url_prefix = url_prefix
self.routes = []
def get(self, path):
def decorator(handler):
self.routes.append(('GET', path, handler))
return handler
return decorator
def post(self, path):
def decorator(handler):
self.routes.append(('POST', path, handler))
return handler
return decorator
def register(self, app):
for method, path, handler in self.routes:
full_path = f'{self.url_prefix}{path}'
app.router.add_route(method.upper(), full_path, handler)
users_bp = RouterGroup('/api/v1/users')
@users_bp.get('/')
async def list_users(request):
return web.json_response([
{'id': 1, 'name': '张三'},
{'id': 2, 'name': '李四'},
])
def setup_groups(app):
users_bp.register(app)
中间件系统
基础中间件
from aiohttp import web
import time
# 方式一：简单中间函数
async def timing_middleware(request, handler):
start = time.monotonic()
response = await handler(request)
elapsed = time.monotonic() - start
print(f"[{request.method}] {request.path} → {response.status} ({elapsed:.3f}s)")
return response
# 方式二：带参数的中间件工厂
def logging_middleware():
async def middleware(request, handler):
before = time.monotonic()
try:
response = await handler(request)
status = response.status
except Exception:
status = 500
raise
duration_ms = (time.monotonic() - before) * 1000
remote = request.remote or 'unknown'
ua = request.headers.get('User-Agent', '-')
print(f"{remote} {request.method} {request.path} {status} {duration_ms:.0f}ms")
return response
return middleware
# 方式三：认证中间件
def auth_middleware(required_roles=None):
required_roles = required_roles or []
async def middleware(request, handler):
if request.path in ('/auth/login', '/auth/register'):
return await handler(request)
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if not token:
return web.json_response({'error': '未提供认证令牌'}, status=401)
return await handler(request)
return middleware
# 方式四：CORS 中间件
async def cors_middleware(request, handler):
response = await handler(request)
if request.method == 'OPTIONS':
response = web.Response(status=204)
response.headers['Access-Control-Allow-Origin'] = '*'
response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS'
response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
return response
def setup_middlewares(app):
app.middlewares.append(timing_middleware)
app.middlewares.append(logging_middleware())
app.middlewares.append(auth_middleware(['admin']))
app.middlewares.append(cors_middleware)
异常中间件
from aiohttp import web
import traceback
import sys
async def error_handling_middleware(request, handler):
try:
return await handler(request)
except web.HTTPException:
raise
except Exception as e:
print(f"Uncaught exception: {e}", file=sys.stderr)
traceback.print_exc()
return web.json_response(
{'error': 'Internal Server Error', 'message': str(e)},
status=500
)
app = web.Application(middlewares=[error_handling_middleware])
缓存中间件
from aiohttp import web
import hashlib
import time
cache_store = {}
CACHE_TTL = 60
def cache_middleware(max_age=CACHE_TTL):
async def middleware(request, handler):
if request.method != 'GET':
return await handler(request)
cache_key = hashlib.md5(request.url.raw_query_string.encode()).hexdigest()
cached = cache_store.get(cache_key)
now = time.time()
if cached and (now - cached['time']) < max_age:
resp = web.Response(body=cached['body'])
resp.status = cached['status']
resp.headers.update(cached['headers'])
return resp
response = await handler(request)
if response.status >= 200 and response.status < 300:
body_bytes = b''
if hasattr(response, 'body') and response.body:
body_bytes = response.body
cache_store[cache_key] = {
'body': body_bytes,
'status': response.status,
'headers': dict(response.headers),
'time': now,
}
response.headers['Cache-Control'] = f'max-age={max_age}'
return response
return middleware
实战：异步 REST API 博客系统
"""
完整异步博客 API — aiohttp 实现
包含：CRUD、分页、搜索、限流、Swagger 文档生成
"""
from aiohttp import web
import json, time, uuid, math, collections
from datetime import datetime
class BlogDB:
"""内存数据库"""
def __init__(self):
self.posts = [
{
'id': str(uuid.uuid4())[:8],
'title': 'Python 异步编程完全指南',
'content': '本文详细介绍 Python 异步编程的核心概念...',
'author': 'admin',
'category': '技术',
'tags': ['Python', 'asyncio', 'Web'],
'views': 1024,
'likes': 128,
'status': 'published',
'created_at': '2026-01-15T10:30:00Z',
},
]
self.comments = []
self.next_comment_id = 1
def get_published_posts(self, page=1, per_page=10, category=None, search=None):
posts = [p for p in self.posts if p['status'] == 'published']
if category:
posts = [p for p in posts if p['category'] == category]
if search:
keyword = search.lower()
posts = [p for p in posts if keyword in p['title'].lower() or keyword in p['content'].lower()]
posts.sort(key=lambda x: x['created_at'], reverse=True)
total = len(posts)
total_pages = math.ceil(total / per_page) if total > 0 else 1
start = (page - 1) * per_page
return posts[start:start+per_page], total, total_pages
def get_post(self, post_id):
return next((p for p in self.posts if p['id'] == post_id), None)
def create_post(self, title, content, author, category='', tags=None):
now = datetime.utcnow().isoformat() + 'Z'
post = {
'id': str(uuid.uuid4())[:8], 'title': title,
'content': content, 'author': author,
'category': category, 'tags': tags or [],
'views': 0, 'likes': 0, 'status': 'draft',
'created_at': now, 'updated_at': now,
}
self.posts.append(post)
return post
def add_comment(self, post_id, content, author='anonymous'):
comment = {
'id': self.next_comment_id, 'post_id': post_id,
'author': author, 'content': content,
'created_at': datetime.utcnow().isoformat() + 'Z',
}
self.next_comment_id += 1
self.comments.append(comment)
return comment
def get_comments(self, post_id):
return [c for c in self.comments if c['post_id'] == post_id]
def increment_views(self, post_id):
for p in self.posts:
if p['id'] == post_id:
p['views'] += 1
return True
return False
db = BlogDB()
async def health_check(request):
return web.json_response({
'status': 'healthy', 'service': 'blog-api',
'version': '1.0.0', 'uptime': time.time(),
})
async def list_posts(request):
page = int(request.query.get('page', 1))
per_page = min(int(request.query.get('per_page', 10)), 50)
category = request.query.get('category')
search = request.query.get('search')
posts, total, total_pages = db.get_published_posts(
page=page, per_page=per_page, category=category, search=search
)
return web.json_response({
'code': 0,
'data': {
'items': posts,
'pagination': {
'page': page, 'per_page': per_page,
'total': total, 'total_pages': total_pages,
'has_next': page < total_pages,
}
}
})
async def get_post_detail(request):
post_id = request.match_info['id']
post = db.get_post(post_id)
if not post:
return web.json_response({'code': 1, 'error': '文章不存在'}, status=404)
await asyncio.sleep(0.001)
db.increment_views(post_id)
comments = db.get_comments(post_id)
return web.json_response({'code': 0, 'data': {**post, 'comments_count': len(comments)}})
async def create_post(request):
data = await request.json()
if not data.get('title'):
return web.json_response({'code': 1, 'error': '标题不能为空'}, status=400)
if not data.get('content'):
return web.json_response({'code': 1, 'error': '内容不能为空'}, status=400)
post = db.create_post(
title=data['title'], content=data['content'],
author=data.get('author', 'anonymous'),
category=data.get('category', ''),
tags=data.get('tags', []),
)
return web.json_response({'code': 0, 'data': post}, status=201)
async def add_comment(request):
post_id = request.match_info['id']
existing = db.get_post(post_id)
if not existing:
return web.json_response({'code': 1, 'error': '文章不存在'}, status=404)
data = await request.json()
if not data.get('content'):
return web.json_response({'code': 1, 'error': '评论内容不能为空'}, status=400)
comment = db.add_comment(post_id=post_id, content=data['content'], author=data.get('author', 'anonymous'))
return web.json_response({'code': 0, 'data': comment}, status=201)
async def get_comments(request):
post_id = request.match_info['id']
comments = db.get_comments(post_id)
return web.json_response({'code': 0, 'data': comments})
class SimpleRateLimiter:
def __init__(self, requests_per_second=10, window_size=60):
self.requests_per_second = requests_per_second
self.window_size = window_size
self._requests = collections.defaultdict(list)
def is_allowed(self, client_ip):
now = time.time()
timestamps = self._requests[client_ip]
self._requests[client_ip] = [t for t in timestamps if now - t < self.window_size]
timestamps = self._requests[client_ip]
if len(timestamps) >= self.requests_per_second:
return False
timestamps.append(now)
return True
rate_limiter = SimpleRateLimiter(requests_per_second=5)
async def rate_limit_middleware(request, handler):
client_ip = request.remote or 'unknown'
if not rate_limiter.is_allowed(client_ip):
resp = web.json_response({'code': 1, 'error': '请求过于频繁'}, status=429)
resp.headers['Retry-After'] = '60'
return resp
return await handler(request)
async def ws_chat(request):
"""WebSocket 聊天室"""
ws = web.WebSocketResponse(heartbeat=30)
await ws.prepare(request)
clients = set()
clients.add(ws)
try:
async for msg in ws:
if msg.type == web.WSMsgType.TEXT:
message = f"[Guest] {msg.data}"
dead = set()
for client in clients:
try:
await client.send_str(message)
except Exception:
dead.add(client)
clients -= dead
elif msg.type == web.WSMsgType.ERROR:
break
finally:
clients.discard(ws)
return ws
def generate_swagger_json():
"""动态生成 OpenAPI 3.0 规范"""
return {
'openapi': '3.0.0',
'info': {'title': 'Blog API', 'description': '异步博客 REST API 文档', 'version': '1.0.0'},
'paths': {
'/api/posts': {
'get': {'summary': '获取文章列表', 'responses': {'200': {'description': '成功'}}},
'post': {'summary': '创建新文章', 'responses': {'201': {'description': '已创建'}}},
},
'/api/posts/{id}': {
'get': {'summary': '获取文章详情'},
},
}
}
async def swagger_json(request):
return web.json_response(generate_swagger_json())
async def swagger_ui(request):
html = '''<!DOCTYPE html><html><head><meta charset="UTF-8">
<title>Blog API Docs</title>
<script src="https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui-bundle.js"></script>
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui.css">
</head><body><div id="swagger-ui"></div>
<script>SwaggerUIBundle({url: "/api/docs.json", dom_id: "#swagger-ui"});</script></body></html>'''
return web.Response(text=html, content_type='text/html')
def create_app():
app = web.Application(middlewares=[rate_limit_middleware, error_handling_middleware, cors_middleware])
app.router.add_get('/health', health_check)
app.router.add_get('/ping', lambda r: web.Response(text='pong'))
app.router.add_get('/api/posts', list_posts)
app.router.add_post('/api/posts', create_post)
app.router.add_get('/api/posts/{id}', get_post_detail)
app.router.add_get('/api/posts/{id}/comments', get_comments)
app.router.add_post('/api/posts/{id}/comments', add_comment)
app.router.add_get('/api/docs.json', swagger_json)
app.router.add_get('/api/docs', swagger_ui)
app.router.add_get('/api/ws/chat', ws_chat)
return app
app = create_app()
if __name__ == '__main__':
print("✓ Blog API 运行于 http://localhost:8080")
print(" 文档: http://localhost:8080/api/docs")
print(" WS: ws://localhost:8080/api/ws/chat")
web.run_app(app, host='0.0.0.0', port=8080)
性能对比
异步 vs 同步基准测试
import asyncio
import aiohttp
import time
async def async_benchmark(base_url, num_requests=50):
"""异步并发请求"""
start = time.perf_counter()
async with aiohttp.ClientSession() as session:
sem = asyncio.Semaphore(20)
async def fetch(url):
async with sem:
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
await resp.text()
return resp.status
except Exception:
return -1
tasks = [fetch(f'{base_url}/api/posts?page={i % 5}') for i in range(num_requests)]
results = await asyncio.gather(*tasks)
elapsed = time.perf_counter() - start
success = sum(1 for r in results if r == 200)
print("=" * 50)
print(f"异步 aiohttp 测试结果:")
print(f" 请求数: {num_requests} | 耗时: {elapsed:.2f}s | QPS: {num_requests/elapsed:.0f}")
print(f" 成功率: {success}/{num_requests}")
print(f" 平均延迟: {(elapsed/num_requests)*1000:.1f}ms")
print("=" * 50)
# 先启动博客 API: python app.py
# 然后运行: asyncio.run(async_benchmark('http://localhost:8080'))
对比结果分析
| 框架 | 模式 | QPS (近似) | 内存占用 | 适用场景 |
|---|---|---|---|---|
| aiohttp | 异步 | 5000+ | ~10MB | 高并发 API |
| Flask | 同步 | 500-1000 | ~50MB | 中小型应用 |
| Django | 同步 | 300-800 | ~100MB | 全栈大型应用 |
注意：实际数值取决于硬件和具体代码。异步框架在 IO 密集型任务下优势明显；CPU 密集型和简单小项目用同步框架更简单高效。
常见问题与调试
如何调试异步代码
import asyncio
async def debug_example():
loop = asyncio.get_event_loop()
print(f"Event Loop: {type(loop).__name__}")
# 查看所有 Task
tasks = asyncio.all_tasks(loop)
for task in tasks:
print(f"Task: {task.get_coro().__name__}")
# 开启调试模式（coroutine 丢弃或超过 100ms 时警告）
asyncio.get_event_loop().set_debug(True)
常见错误速查
| 错误信息 | 原因 | 解决方案 |
|---|---|---|
RuntimeError: This event loop is already running |
嵌套调用 asyncio.run() |
只在一个顶层调用 |
RuntimeWarning: coroutine was never awaited |
忘记 await | 确保每个 async 函数都加上 await |
ResourceWarning: unclosed transport |
资源泄漏 | 使用 async with 关闭 Session |
ConnectionResetError |
客户端断开连接 | try-except 包裹处理 |
小结
| 知识点 | 说明 |
|---|---|
| web.run_app() | 最简单的启动方式 |
| add_get/add_post/... | 路由注册方法 |
| request.match_info | 路径参数 {id} |
| request.query | Query String ?key=value |
| request.json/text/read | 不同格式的 Body 读取 |
| web.json_response | 自动序列化的 JSON 响应 |
| 中间件列表 | 按注册顺序执行（洋葱模型） |
| CORS 中间件 | 跨域资源共享支持 |
| Rate Limiter | 简易 IP 限流 |
| WebSocket | web.WebSocketResponse 实时通信 |
| add_static | 静态文件服务 |
| OpenAPI/Swagger | 动态生成 API 文档 |
核心理解：aiohttp 让你用熟悉的路由器和 handler 模式编写异步代码，同时享受事件循环带来的高性能。学习 aiohttp 后，你也会更容易理解 Starlette、FastAPI 等其他异步框架——它们底层都在使用类似的事件循环和协议抽象。
练习题
为博客 API 添加分页排序功能：支持
sort_by参数（views/likes/created_at），支持order参数（asc/desc），并在分页信息中返回上一页/下一页 URL。实现一个WebSocket 群聊机器人：当有用户发送
/bot ping消息时，机器人自动回复pong（附带时间戳）；每在线人数变化时，向所有客户端广播一条通知消息。对比测试：在同一台机器上，分别用 aiohttp 和 Flask 搭建相同逻辑的博客 API，各发起 100 个并发请求，记录 QPS、平均延迟和 P99 延迟，分析差异原因。