WSGI 与 ASGI 是什么?
**WSGI(Web Server Gateway Interface)和ASGI(Asynchronous Server Gateway Interface)**是定义 Web 服务器如何与 Python Web 应用程序和框架通信的规范。
可以把它们想象成"合同"或"协议",确保 Web 服务器和 Python 应用能无缝协作。
WSGI(Web 服务器网关接口)
概述
o 创建时间:2003 年(PEP 333,后在 PEP 3333 更新)
o 目的:标准化 Web 服务器与 Python Web 应用之间的接口
o 模型:同步,阻塞 I/O
o 状态:成熟,广泛采用的标准
WSGI 如何工作
WSGI 定义了一个简单的双边接口:
1. 服务器端:实现 WSGI 的 Web 服务器
2. 应用端:符合 WSGI 的 Python 应用
WSGI 应用结构
WSGI 应用本质上是一个可调用对象(函数或类),它:
o 接受两个参数:environ 和 start_response
o 返回一个字节字符串的可迭代对象
def simple_wsgi_app(environ, start_response):
"""
environ: 包含类 CGI 环境变量的字典
start_response: 初始化 HTTP 响应的可调用对象
"""
status = '200 OK'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
return [b'Hello, WSGI World!']
# 基于类的 WSGI 应用
class WSGIApp:
def __call__(self, environ, start_response):
status = '200 OK'
headers = [('Content-Type', 'text/html')]
start_response(status, headers)
path = environ.get('PATH_INFO', '/')
method = environ.get('REQUEST_METHOD', 'GET')
response = f"""
<html>
<body>
<h1>WSGI 应用</h1>
<p>路径: {path}</p>
<p>方法: {method}</p>
</body>
</html>
""".encode('utf-8')
return [response]
app = WSGIApp()
WSGI 服务器
常见的 WSGI 服务器包括:
o Gunicorn - Unix 系统的 Python WSGI HTTP 服务器
o uWSGI - 全功能应用服务器
o mod_wsgi - Apache 模块
o Waitress - 生产级纯 Python WSGI 服务器
o Werkzeug - 开发服务器(Flask 使用)
WSGI 框架
o Django - 全功能 Web 框架
o Flask - 轻量级微框架
o Bottle - 极简框架
o Pyramid - 灵活框架
使用 Flask 的 WSGI 示例
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello():
return 'Hello, WSGI with Flask!'
@app.route('/user/<name>')
def user(name):
return f'Hello, {name}!'
# 这是一个 WSGI 应用
# Flask 自动创建 WSGI 接口
if __name__ == '__main__':
# 开发服务器
app.run(debug=True)
# 生产环境使用 WSGI 服务器:
# gunicorn app:app
ASGI(异步服务器网关接口)
概述
o 创建时间:2016 年
o 目的:扩展 WSGI 以支持异步 Python 和处理 WebSocket、HTTP/2 等
o 模型:异步,非阻塞 I/O
o 状态:现代异步 Python Web 应用的标准
为什么需要 ASGI?
WSGI 的局限:
o 仅同步 - 阻塞 I/O 操作
o 仅 HTTP - 不能处理 WebSocket、HTTP/2 服务器推送
o 请求-响应循环 - 不支持长连接
o 线程开销 - 每个请求通常需要一个线程
ASGI 的解决方案:
o 异步 - 使用 async/await 的非阻塞 I/O
o 协议无关 - 支持 HTTP、WebSocket、HTTP/2 等
o 长连接 - 支持持久连接
o 更好的性能 - 单线程事件循环
ASGI 应用结构
ASGI 应用是一个异步可调用对象,它:
o 接受三个参数:scope、receive、send
o 使用 async/await 语法
async def simple_asgi_app(scope, receive, send):
"""
scope: 包含连接信息的字典
receive: 接收消息的异步可调用对象
send: 发送消息的异步可调用对象
"""
if 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, ASGI World!',
})
# 基于类的 ASGI 应用
class ASGIApp:
async def __call__(self, scope, receive, send):
if scope['type'] == 'http':
await self.handle_http(scope, receive, send)
elif scope['type'] == 'websocket':
await self.handle_websocket(scope, receive, send)
async def handle_http(self, scope, receive, send):
path = scope.get('path', '/')
method = scope.get('method', 'GET')
response_body = f"""
<html>
<body>
<h1>ASGI 应用</h1>
<p>路径: {path}</p>
<p>方法: {method}</p>
</body>
</html>
""".encode('utf-8')
await send({
'type': 'http.response.start',
'status': 200,
'headers': [[b'content-type', b'text/html']],
})
await send({
'type': 'http.response.body',
'body': response_body,
})
async def handle_websocket(self, scope, receive, send):
await send({'type': 'websocket.accept'})
while True:
message = await receive()
if message['type'] == 'websocket.receive':
await send({
'type': 'websocket.send',
'text': f"Echo: {message.get('text', '')}"
})
elif message['type'] == 'websocket.disconnect':
break
app = ASGIApp()
ASGI 服务器
常见的 ASGI 服务器包括:
o Uvicorn - 闪电般快速的 ASGI 服务器
o Hypercorn - 支持 HTTP/2 和 HTTP/3
o Daphne - HTTP、HTTP/2 和 WebSocket 协议服务器
o Gunicorn - 配合 uvicorn worker 类
ASGI 框架
o FastAPI - 现代、高性能 API 框架
o Starlette - 轻量级 ASGI 框架
o Django - 从 Django 3.0 开始支持 ASGI
o Quart - Flask 的异步版本
o Sanic - 异步 Web 框架
使用 FastAPI 的 ASGI 示例
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
import asyncio
import aiofiles
app = FastAPI()
@app.get("/")
async def read_root():
return {"Hello": "ASGI World"}
@app.get("/async-file")
async def read_file():
# 非阻塞文件 I/O
async with aiofiles.open('data.txt', mode='r') as f:
content = await f.read()
return {"content": content}
@app.get("/slow-endpoint")
async def slow_endpoint():
# 非阻塞睡眠 - 其他请求可以继续处理
await asyncio.sleep(5)
return {"message": "这花了5秒但没有阻塞其他请求"}
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"收到消息: {data}")
# 运行:uvicorn main:app --reload
关键差异对比
方面WSGIASGI执行模型同步,阻塞异步,非阻塞并发线程/多进程单线程事件循环协议仅 HTTPHTTP、WebSocket、HTTP/2 等连接类型仅请求-响应支持长连接I/O 操作阻塞非阻塞内存使用较高(每请求一个线程)较低(共享事件循环)性能CPU 密集型任务好I/O 密集型任务好复杂度较简单较复杂成熟度非常成熟(20+年)较新(8+年)
性能对比
WSGI 性能特征
# WSGI - 每个请求都会阻塞直到完成
import time
import requests
def wsgi_blocking_operation():
# 这会阻塞整个线程
time.sleep(1) # 模拟数据库查询
return "数据库数据"
@app.route('/data')
def get_data():
result = wsgi_blocking_operation()
return result
# 10个并发请求:
# - WSGI 需要10个线程
# - 总时间:~1秒(有足够线程时)
# - 内存使用:高(线程开销)
ASGI 性能特征
# ASGI - 并发请求互不阻塞
import asyncio
import aiohttp
async def asgi_async_operation():
# 不会阻塞事件循环
await asyncio.sleep(1) # 模拟异步数据库查询
return "数据库数据"
@app.get('/data')
async def get_data():
result = await asgi_async_operation()
return result
# 10个并发请求:
# - ASGI 使用单个事件循环
# - 总时间:~1秒(所有请求并发处理)
# - 内存使用:低(无线程开销)
何时使用哪个
使用 WSGI 的场景:
o 构建传统 Web 应用
o 使用成熟框架(Django、Flask)
o CPU 密集型操作
o 简单的请求-响应模式
o 团队熟悉同步编程
o 现有基础设施基于 WSGI
使用 ASGI 的场景:
o 构建现代 API 或微服务
o 需要 WebSocket 支持
o 高 I/O 操作(数据库、文件系统、网络调用)
o 实时应用
o 需要并发请求的最大性能
o 从头开始构建新应用
迁移考虑
WSGI 到 ASGI 迁移
# WSGI Flask 应用
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello():
return 'Hello World'
# 转换为使用 Quart 的 ASGI
from quart import Quart
app = Quart(__name__)
@app.route('/')
async def hello():
return 'Hello World'
# 或使用 ASGI 适配器
from asgiref.wsgi import WsgiToAsgi
from flask import Flask
flask_app = Flask(__name__)
app = WsgiToAsgi(flask_app)
同时运行两者
# 可以使用适配器在 ASGI 服务器上运行 WSGI 应用
from fastapi import FastAPI
from fastapi.middleware.wsgi import WSGIMiddleware
from flask import Flask
# WSGI Flask 应用
flask_app = Flask(__name__)
@flask_app.route('/legacy')
def legacy_endpoint():
return '这是一个传统的 WSGI 端点'
# ASGI FastAPI 应用
fastapi_app = FastAPI()
@fastapi_app.get('/modern')
async def modern_endpoint():
return {'message': '这是一个现代的 ASGI 端点'}
# 在 ASGI 应用中挂载 WSGI 应用
fastapi_app.mount('/wsgi', WSGIMiddleware(flask_app))
结论
WSGI 仍然非常适合传统 Web 应用,特别是在使用成熟框架和团队熟悉同步编程的情况下。
ASGI 是高性能、现代 Web 应用的未来,特别是需要处理大量并发连接、实时功能或密集 I/O 操作的场景。
许多组织正在逐步从 WSGI 迁移到 ASGI,或在过渡期使用混合方法让两者共存。