使用aiohttp

构建异步Web服务入门

为什么选择它？

asyncio是Python内置的单线程并发I/O框架，而aiohttp是其生态下非常成熟的异步HTTP客户端+服务器双端库。相比Flask/Django等同步框架，它不需要多进程/线程就能轻松处理数千个并发连接，内存占用也更低，原生支持WebSocket这类长连接场景，非常适合API网关、实时推送、爬虫服务这类I/O密集型应用。

环境准备

  1. 基础要求：Python 3.7+（asyncio在3.7后引入更简洁的async/await语法增强和更稳定的事件循环实现）
  2. 安装最新版库：
pip install aiohttp

第一个异步Web服务器

先实现一个极简但能用的双路由服务器，包含静态首页和动态参数路由：

  1. / → 返回带HTML标题的首页
  2. /{nickname} → 根据URL昵称返回个性化问候
# app.py
from aiohttp import web

async def handle_index(request):
    """根路由异步处理函数：返回静态首页"""
    html_content = "<h1>✨ aiohttp 异步首页 ✨</h1><p>访问 /你的昵称 试试吧！</p>"
    # 指定text/html类型，避免浏览器直接展示纯文本
    return web.Response(text=html_content, content_type="text/html")

async def handle_hello(request):
    """动态昵称路由处理函数：获取URL参数"""
    # match_info从URL占位符提取值，没匹配到给默认值World
    nickname = request.match_info.get("nickname", "陌生访客")
    html_content = f"<h1>👋 Hello, {nickname}!</h1>"
    return web.Response(text=html_content, content_type="text/html")

def make_app():
    """应用工厂函数：创建实例并绑定路由"""
    app = web.Application()
    # 批量添加GET请求的路由
    app.add_routes([
        web.get("/", handle_index),
        web.get("/{nickname}", handle_hello)
    ])
    return app

if __name__ == "__main__":
    # 启动本地开发服务器，默认监听0.0.0.0:8080（局域网也能访问）
    app = make_app()
    web.run_app(app, host="0.0.0.0")

代码细节拆解

1. 异步处理函数

  • 必须用async def定义，这样事件循环才能调度它
  • 强制接收web.Request对象作为唯一参数，里面封装了所有请求信息：URL、请求头、Body、Cookie等
  • 必须返回web.Response的子类（如json_response、FileResponse），不能直接返回字符串/字典

2. 路由配置

  • 用web.Application()初始化服务核心实例
  • 路由支持多种HTTP方法：web.get(), web.post(), web.put(), web.delete(), web.patch()
  • 批量添加用add_routes()，单条添加用app.router.add_*()
  • URL占位符支持简单的正则约束，比如/{nickname:\w+}只匹配字母数字下划线

3. 应用工厂函数

示例中的make_app()不是多余的，生产中常用：

  • 方便为测试/开发/生产环境创建不同配置的实例
  • 避免全局变量污染
  • 更容易与部署工具（如Gunicorn+uvloop）集成

运行与快速测试

1. 本地启动

python app.py

控制台会输出：

======== Running on http://0.0.0.0:8080 ========
(Press CTRL+C to quit)

2. 浏览器/命令行访问

  • 首页：直接打开浏览器输入http://localhost:8080/
  • 动态页：输入http://localhost:8080/Alice
  • 命令行用curl测试：
# 首页
curl http://localhost:8080/
# 动态页
curl http://localhost:8080/Bob

几个常用的高级配置

自定义监听端口和地址

if __name__ == "__main__":
    # 只允许本机访问，端口改为9000
    web.run_app(app, host="127.0.0.1", port=9000)

快速返回JSON响应

aiohttp内置了json_response，不需要手动设置Content-Type和序列化：

from aiohttp import web

async def handle_api(request):
    data = {"code": 200, "msg": "请求成功", "data": {"nickname": "Charlie"}}
    return web.json_response(data)

添加中间件

中间件可以处理所有请求的前置/后置逻辑，比如统一日志、鉴权、跨域设置：

from aiohttp import web

async def log_middleware(request, handler):
    # 前置：记录请求方法和路径
    print(f"[REQUEST] {request.method} {request.path}")
    # 调用下一个中间件/处理函数
    response = await handler(request)
    # 后置：记录响应状态码
    print(f"[RESPONSE] {request.method} {request.path} → {response.status}")
    return response

def make_app():
    # 初始化时传入中间件列表
    app = web.Application(middlewares=[log_middleware])
    app.add_routes([
        web.get("/", handle_index),
        web.get("/{nickname}", handle_hello),
        web.get("/api", handle_api)
    ])
    return app

生产环境部署建议

  1. 不要用web.run_app()直接部署（它是开发服务器，性能和稳定性有限）
  2. 推荐搭配Gunicorn+uvloop：
# 先安装uvloop（替代默认事件循环，性能提升2-4倍）
pip install uvloop gunicorn
# 启动4个worker进程（每个worker是独立事件循环，绑定到0.0.0.0:9000）
gunicorn app:make_app --bind 0.0.0.0:9000 --worker-class aiohttp.GunicornUVLoopWebWorker --workers 4
  1. 前面加Nginx做反向代理：处理静态文件、负载均衡、SSL证书配置

总结

本教程快速入门了aiohttp的服务器端用法，包括基础路由、异步处理、JSON响应和中间件。如果需要更复杂的功能，aiohttp还提供了模板渲染（需配合aiohttp-jinja2）、Session管理（需配合aiohttp-session）、WebSocket长连接等扩展，非常适合生产级别的I/O密集型Web应用。