本文详解如何在 fastapi 中正确设计 websocket 连接管理器，支持同一用户（customer_id）在多个设备（如手机、pc）上同时建立独立连接，避免因重复连接拒绝导致的前端重试风暴。
本文详解如何在 fastapi 中正确设计 websocket 连接管理器，支持同一用户（customer_id）在多个设备（如手机、pc）上同时建立独立连接，避免因重复连接拒绝导致的前端重试风暴。
在实际生产场景中，用户常需跨设备使用同一账号——例如桌面浏览器与移动 App 同时在线接收实时通知。但原始实现仅以 customer_id 为键做单连接校验，导致新设备连接时强制关闭旧连接或直接拒绝，进而触发浏览器反复重连（如 Firefox 日志中连续 3 次 403），形成“连接雪崩”。
根本问题在于：连接粒度过粗。应将连接标识从“用户级”升级为“用户 × 设备级”，即每个设备拥有唯一标识（device_hash），从而允许多连接共存。
✅ 正确方案：二维连接映射结构
我们重构 ConnectionManager，采用嵌套字典 clients: Dict[int, Dict[str, WebSocket]]，外层以 customer_id 分组，内层以 device_hash 区分终端：
from typing import Dict, Optional
from fastapi import WebSocket, WebSocketDisconnect
import hashlib
class ExistingConnectionError(Exception):
pass
class ConnectionManager:
def __init__(self):
# {customer_id: {device_hash: websocket}}
self.clients: Dict[int, Dict[str, WebSocket]] = {}
async def connect(self, websocket: WebSocket, customer_id: int, device_hash: str):
# 获取该用户的设备连接池
client_connections = self.clients.setdefault(customer_id, {})
# 若该设备已存在活跃连接，拒绝新连接（可选：也可主动踢下线）
if device_hash in client_connections:
await websocket.close(code=4001, reason="Device already connected")
raise ExistingConnectionError(f"Connection exists for device {device_hash}")
await websocket.accept()
websocket.customer_id = customer_id
websocket.device_hash = device_hash
client_connections[device_hash] = websocket
print(f"[Connected] customer={customer_id}, device={device_hash}")
async def disconnect(self, websocket: WebSocket):
customer_id = getattr(websocket, "customer_id", None)
device_hash = getattr(websocket, "device_hash", None)
if customer_id and device_hash:
client_connections = self.clients.get(customer_id)
if client_connections and device_hash in client_connections:
client_connections.pop(device_hash)
# 清理空用户组
if not client_connections:
self.clients.pop(customer_id)
print(f"[Disconnected] customer={customer_id}, device={device_hash}")
async def broadcast_to_customer(self, customer_id: int, message: dict):
client_connections = self.clients.get(customer_id)
if client_connections:
for ws in client_connections.values():
try:
await ws.send_json(message)
except RuntimeError:
# 连接已关闭，跳过
continue
async def broadcast_to_all(self, message: dict):
for client_connections in self.clients.values():
for ws in client_connections.values():
try:
await ws.send_json(message)
except RuntimeError:
continue
connection_manager = ConnectionManager()? 前端适配：生成稳定 device_hash
device_hash 应具备设备稳定性（同设备每次连接一致）和跨平台可生成性。推荐两种方式：
WebSocket 8.18.2 是该协议规范的一个重要迭代版本，主要优化了连接稳定性与数据传输效率。它通过全双工通信机制，允许客户端与服务器在单一长连接上实时交换数据，大幅降低传统 HTTP 轮询的开销。该版本增强了心跳保活、自动重连及二进制帧传输能力，适用于即时通讯、在线游戏及金融行情推送等低延迟场景，为开发者提供更可靠的实时网络交互基础。
方式一：客户端生成（推荐）
利用 navigator.userAgent + screen.width + screen.height + navigator.hardwareConcurrency 等指纹字段哈希：
function generateDeviceHash(): string {
const fingerprint = `${navigator.userAgent}-${screen.width}x${screen.height}-${navigator.hardwareConcurrency}`;
// 简易哈希（生产环境建议用 crypto.subtle.digest）
return btoa(fingerprint).substring(0, 16);
}
const deviceHash = generateDeviceHash();
const socket = new WebSocket(`ws://localhost:3002/sock?customer_id=${customerId}&device_hash=${deviceHash}`);方式二：服务端推导（免前端改造）
在 FastAPI 路由中解析 User-Agent 等 Header 自动生成（注意：需确保 header 可靠性）：
@app.websocket("/sock")
async def websocket_endpoint(
websocket: WebSocket,
customer_id: int,
device_hash: Optional[str] = None,
user_agent: str = Header(default=""),
):
# 若未传 device_hash，服务端生成（仅作兜底）
if not device_hash:
device_hash = hashlib.md5(user_agent.encode()).hexdigest()[:16]
try:
await connection_manager.connect(websocket, customer_id, device_hash)
while True:
data = await websocket.receive_json()
# 处理业务逻辑，如转发给其他设备
await connection_manager.broadcast_to_customer(customer_id, {"type": "echo", "data": data})
except WebSocketDisconnect:
await connection_manager.disconnect(websocket)
except ExistingConnectionError as e:
print(str(e))⚠️ 关键注意事项
-
不要依赖
WebSocket.close()后立即清理状态：disconnect需在except WebSocketDisconnect中调用，且应捕获RuntimeError（连接已断）避免异常中断。 -
避免内存泄漏：务必在
disconnect中清理嵌套字典，并移除空用户项（见代码中if not client_connections: pop）。 -
错误码语义化：使用标准 WebSocket 错误码（如
4001表示“duplicate device”），便于前端区分处理。 -
广播性能优化：高频广播场景建议引入异步任务队列（如
asyncio.create_task()）或消息中间件解耦。
✅ 总结
通过引入 device_hash 维度，我们将连接模型从「单用户单连接」升级为「单用户多设备多连接」，既满足现代多端协同需求，又保持后端逻辑清晰可控。该模式可无缝扩展至会话管理、设备在线状态同步、跨端指令下发等高级场景。
