ASGI 是异步服务器网关接口（Asynchronous Server Gateway Interface）的缩写，是 WSGI 的异步进化版本，专为现代 Python 异步 Web 框架设计。

基本概念 #

1. ASGI 是什么 #

ASGI 是一个介于 Python Web 应用和服务器之间的标准接口，类似于 WSGI，但支持异步操作。它允许：

• 异步请求处理
• 长连接（如 WebSockets）
• 协议无关（HTTP/1.1, HTTP/2, WebSocket 等）

2. ASGI 与 WSGI 对比 #

ASGI 规范 #

1. 应用接口 #

# 定义一个最简单的 ASGI 应用
# scope: 包含请求信息的字典
# receive: 接收消息的异步函数
# send: 发送消息的异步函数
async def app(scope, receive, send):
    """
    scope: 包含请求信息的字典
    receive: 接收消息的异步函数
    send: 发送消息的异步函数
    """
    # 这里只是一个接口定义示例，实际运行需要补充完整的 send 逻辑
    pass

2. 消息流 #

1. 连接阶段：服务器创建连接，发送初始信息
2. 交互阶段：应用和服务器通过 receive/send 交换消息
3. 断开阶段：连接关闭

3. 消息类型 #

HTTP 请求示例 #

# 下面是 ASGI HTTP 消息的结构示例，不是可运行代码
# 实际开发中应通过 send/receive 进行消息交互
# 接收到的消息
http_request = {
    "type": "http.request",
    "body": b"request body data",
    "more_body": False
}

# 发送的消息
http_response_start = {
    "type": "http.response.start",
    "status": 200,
    "headers": [(b"content-type", b"text/plain")]
}
http_response_body = {
    "type": "http.response.body",
    "body": b"response body data",
    "more_body": False
}

WebSocket 示例 #

# 下面是 ASGI WebSocket 消息的结构示例，不是可运行代码
# 实际开发中应通过 send/receive 进行消息交互
# 接收到的消息
websocket_connect = {"type": "websocket.connect"}
websocket_receive = {"type": "websocket.receive", "text": "Hello world!"}

# 发送的消息
websocket_accept = {"type": "websocket.accept"}
websocket_send = {"type": "websocket.send", "text": "Message received"}

ASGI 生态系统 #

1. 服务器实现 #

• Uvicorn：最快的 ASGI 服务器之一（基于 uvloop）
• Daphne：Django 官方推荐的 ASGI 服务器
• Hypercorn：支持 HTTP/2 的 ASGI 服务器
• uvloop：ASGI 服务器的高性能事件循环实现

2. 框架支持 #

• FastAPI：现代高性能 API 框架
• Starlette：轻量级 ASGI 框架（FastAPI 的基础）
• Django Channels：为 Django 添加 ASGI 支持
• Quart：类似 Flask 的异步框架

3. 中间件 #

# 定义 ASGI 中间件的写法
# app: 被包装的 ASGI 应用
async def middleware(app):
    # 返回一个新的 ASGI 应用
    async def wrapped_app(scope, receive, send):
        # 预处理逻辑可以放在这里
        # 调用原始应用
        await app(scope, receive, send)
        # 后处理逻辑可以放在这里
    return wrapped_app

常见中间件：

• 认证
• CORS
• GZip 压缩
• 请求/响应修改

ASGI 工作原理 #

1. 请求生命周期 #

1. 客户端发起请求
2. ASGI 服务器接收并解析请求
3. 服务器创建 scope 字典并调用应用
4. 应用通过 receive 获取请求体
5. 应用通过 send 发送响应
6. 服务器将响应发送给客户端

2. 异步处理流程 #

开发 ASGI 应用 #

1. 简单示例 #

# 一个最简单的 ASGI HTTP 应用，返回 Hello, world!
async def app(scope, receive, send):
    # 确认请求类型为 http
    assert scope['type'] == 'http'
    # 发送响应头
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [
            [b'content-type', b'text/plain'],
        ]
    })
    # 发送响应体
    await send({
        'type': 'http.response.body',
        'body': b'Hello, world!',
    })

2. 使用 Starlette 框架 #

# 导入 Starlette 应用和响应类型
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse

# 创建 Starlette 应用实例
app = Starlette()

# 定义路由和异步视图函数
@app.route('/')
async def homepage(request):
    # 返回纯文本响应
    return PlainTextResponse('Hello, world!')

生产部署 #

1. 服务器选择 #

• 独立运行：Uvicorn 适合轻量级部署
• 多进程：Gunicorn + Uvicorn Worker 适合生产环境
• 容器化：Docker + Uvicorn 适合云部署

2. 性能优化 #

1. 调整工作进程数（通常 CPU 核心数 × 2 + 1）
2. 使用 uvloop 事件循环
3. 启用 keep-alive
4. 使用中间件进行缓存

常见问题 #

1. 同步代码阻塞 #

# 错误示例：同步代码会阻塞事件循环
def sync_function():
    # 阻塞操作
    import time
    time.sleep(1)

# 正确示例：使用异步替代
import asyncio
async def async_function():
    # 非阻塞操作
    await asyncio.sleep(1)

2. 数据库连接 #

# 导入 FastAPI 和异步数据库库
from fastapi import FastAPI
from databases import Database

# 创建 FastAPI 应用实例
app = FastAPI()
# 创建异步数据库对象
# 请将数据库连接字符串替换为实际可用的数据库
database = Database("postgresql://user:password@localhost/db")

# 应用启动时连接数据库
@app.on_event("startup")
async def startup():
    await database.connect()

总结 #

ASGI 代表了 Python Web 开发的未来方向：

• 原生支持异步/等待语法
• 更好的性能表现
• 更灵活的协议支持
• 适合现代 Web 应用需求（如实时通信）

随着 FastAPI、Starlette 等框架的流行，ASGI 正在成为 Python Web 开发的新标准。