Litestar WebSockets 入门指南
Litestar WebSocket 实践：listener 收发消息、显式验证 JSON、stream 为单连接推送，以及认证和跨进程广播的额外要求。
直接回答：Litestar 的 WebSocket 支持与众不同：提供多种实现模式覆盖不同场景——双向消息通信用 listener，单连接数据推送可用 stream；多客户端广播需连接管理或 ChannelsPlugin，JSON 入站校验仍需显式设计。
Litestar WebSocket 的设计理念
多数框架只给你一个"连接对象"自由发挥，Litestar 把常见模式做成了框架能力：事件驱动的 listener 和流式推送，并可结合独立的频道插件管理广播。消息可解析及序列化，但 listener 的入站 JSON 并不自动执行模型字段校验。
项目搭建
pip install 'litestar[standard]' pydantic
WebSocket Listener：事件驱动
from litestar import Litestar, WebSocket
from litestar.handlers import websocket_listener
@websocket_listener("/ws/echo")
async def echo(socket: WebSocket, data: str) -> str:
return f"回声: {data}"
app = Litestar([echo])
listener 模式下每条入站消息触发一次 handler，返回值自动发回客户端——写起来像普通函数，连接管理框架包办。
前端：
const ws = new WebSocket("ws://localhost:8000/ws/echo");
ws.onmessage = (e) => console.log(e.data);
ws.onopen = () => ws.send("你好");
类型安全的消息
from pydantic import BaseModel, Field, ValidationError
class ChatMessage(BaseModel):
user: str = Field(min_length=1, max_length=80)
text: str = Field(min_length=1, max_length=2000)
@websocket_listener("/ws/chat")
async def chat(socket: WebSocket, data: str) -> dict:
try:
message = ChatMessage.model_validate_json(data)
except ValidationError:
return {"error": "消息格式无效"}
return {"user": message.user, "text": message.text.upper()}
# 注册所有处理器后再构造应用
app = Litestar([echo, chat])
这里显式使用 Pydantic 校验原始文本。官方文档说明 listener 的 JSON 数据仅解析、不自动校验；生产用户名须取认证上下文，不能信任客户端 user 字段。
广播与多客户端
简单广播：维护连接集合，收到消息扇出。Litestar 更进一步提供流式抽象与跨进程支持（通过 Redis 等后端频道），多副本部署时广播消息能到达任意副本上的连接——需配置 ChannelsPlugin 和共享后端，并设计慢消费者及失败处理，stream 本身不等于广播。
生命周期钩子
listener 装饰器内可注册 on_accept / on_disconnect：on_accept 在接收连接后执行；若需握手前拒绝，应使用守卫或底层 websocket 处理器、断开时从房间名单移除。优先使用安全会话或短期票据，校验 Origin；首条应用消息在握手后发送，不是握手本身。
生产要点
- 分别记录活跃连接数、消息速率和异常断开数；广播还应记录扇出失败与慢消费者，避免只看握手成功率；
- 广播风暴防护：房间人数上限、消息频率限制；
- 代理支持 WebSocket 升级并配置超时；多进程广播需共享后端，会话保持不能替代它。
常见问题（FAQ）
Q：Litestar WebSocket 和 FastAPI 的有什么区别？
A：FastAPI 提供连接对象与依赖注入，自由但要自己搭管理器；Litestar 的 listener/stream 模式把常见结构框架化，仍需核对消息解析、验证和生命周期的具体边界，不能概括为所有消息都有更严格校验。
Q：支持 Socket.IO 吗？
A：Litestar 走原生 WebSocket 路线；需要 Socket.IO 的自动重连与降级，可搭配 python-socketio（ASGI 挂载）。
Q：心跳保活要自己做吗？
A：建议做。浏览器原生 API 不能直接发送协议级 ping；可用应用级心跳消息，并配置服务器的 ping/超时——穿透 NAT 与代理的空闲连接回收策略全靠心跳维持。
官方参考
本文基于官方文档整理，未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全；局部片段不等同于完整生产应用。
