Sanic WebSockets 入门指南
Sanic 是为速度而生的 Python 异步 Web 框架，WebSocket 支持内置。本文讲解环境搭建、WebSocket 处理器、客户端界面与测试方法，实现实时消息功能。
直接回答：Sanic 是专注性能的异步 Python Web 框架，WebSocket 是一等公民：@app.websocket 一个装饰器即可建立实时通道，配合原生异步生态支撑即时消息、实时通知与动态内容更新。
Sanic 的位置
Flask 太同步、Django 太全，FastAPI 专注 API——Sanic 的定位是"全功能但全异步"：提供路由、中间件和模块化路由分组，支持 async/await；实际性能需按工作负载验证。
环境搭建
pip install sanic
WebSocket 处理器
from sanic import Sanic
app = Sanic("chat")
@app.websocket("/ws")
async def ws_handler(request, ws):
async for msg in ws:
await ws.send(f"回声: {msg}")
处理器拿到 ws 对象，async for 持续接收、send 随时发送——异步迭代器让"读循环"优雅至极。正常关闭时可结束迭代，异常关闭或发送失败仍需处理，并通过 finally 释放连接相关资源。异步迭代接口需 Sanic 22.9 或更新版本。
广播给多客户端
import asyncio
from websockets.exceptions import ConnectionClosed
clients = set()
@app.websocket("/ws/chat")
async def chat(request, ws):
clients.add(ws)
try:
async for msg in ws:
for c in tuple(clients):
try:
await asyncio.wait_for(c.send(msg), timeout=5)
except (ConnectionClosed, TimeoutError):
clients.discard(c)
await c.close()
finally:
clients.discard(ws)
finally 里移除连接是关键——忘记清理的集合迟早泄漏成内存黑洞。多副本部署时广播要换 Redis Pub/Sub 中转（内存集合只覆盖本进程）。
客户端界面
Sanic 模板或静态文件直接出测试页面：
以下页面保存为 templates/index.html，与服务同源提供；仅用于无敏感数据的本地示例。
<form id="f"><input id="m"><button id="send" disabled>发送</button></form>
<ul id="log"></ul>
<script>
const protocol = location.protocol === "https:" ? "wss:" : "ws:";
const ws = new WebSocket(`${protocol}//${location.host}/ws/chat`);
ws.onopen = () => { document.querySelector("#send").disabled = false; };
ws.onclose = () => { document.querySelector("#send").disabled = true; };
ws.onmessage = (e) => {
const li = document.createElement("li");
li.textContent = e.data;
document.querySelector("#log").append(li);
};
document.querySelector("#f").onsubmit = (e) => {
e.preventDefault();
if (ws.readyState === WebSocket.OPEN) {
ws.send(document.querySelector("#m").value);
}
};
</script>
注册 app.static("/", "./templates/index.html") 后，用 sanic app:app --host 127.0.0.1 --port 8000 启动（代码保存在 app.py）。生产还需鉴权、Origin 允许列表、消息大小/速率限制与有界发送队列。示例逐个发送会受慢客户端影响，不是生产广播架构。
测试 WebSocket
安装 sanic-testing 后可使用测试客户端。下面的 app fixture 应返回上面的应用，这是测试示例，不是已执行的测试结果：
def test_ws(app):
async def exchange(client):
await client.send("hello")
assert await client.recv() == "回声: hello"
_, response = app.test_client.websocket("/ws", mimic=exchange)
assert response.opened is True
也可以用 websockets 库写独立测试客户端连真实端口。核心断言：握手成功、发一条收一条、断开不抛异常。
常见问题（FAQ）
Q：Sanic 和 FastAPI 的 WebSocket 怎么选？
A：主站已是 Sanic 就原地用；新项目两者都行——FastAPI 生态大，Sanic 的异步原生与性能口碑老到。功能层面都够用。
Q：连接数上限大概多少？
A：没有可通用承诺的连接数。应测试消息大小、频率、TLS、文件描述符、内存和慢客户端下的容量；跨进程广播需要共享消息系统。普通 WebSocket 连接建立后已固定到后端，不应把粘性会话视为所有部署的必需条件。
Q：生产部署注意什么？
A：sanic app:app --workers N 多进程；前面架反代终结 TLS；心跳保活穿透代理空闲回收；连接数与消息速率埋点进监控。
官方参考
本文基于官方文档整理，未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全；局部片段不等同于完整生产应用。
