FastAPI WebSocket 多设备支持：实现单用户多终端并发连接-js教程-PHP中文网

FastAPI WebSocket 多设备支持：实现单用户多终端并发连接

本文详解如何在 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

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()）或消息中间件解耦。