FastAPI WebSocket单客户端多连接问题及多端连接限制求助

一、前端重复创建连接的修复 #

当前前端代码里的socket变量如果处于局部作用域（比如组件渲染函数内），会导致每次执行代码都重新创建WebSocket实例，发起多个连接请求。

修改后的前端代码 #

// 将socket变量放在全局或组件持久化作用域（比如React的useRef、Vue的ref）
let socket;

function initWebSocket(customer_id) {
  // 检查连接是否已活跃，避免重复创建
  if (socket?.readyState === WebSocket.OPEN) {
    console.log('WebSocket连接已存在，无需重复创建');
    return;
  }

  // 关闭可能存在的失效连接
  if (socket) {
    socket.close();
  }

  socket = new WebSocket(`ws://localhost:3002/sock?customer_id=${customer_id}`);

  socket.onopen = () => {
    console.log('Connected to WebSocket server');
  };

  socket.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log('Incoming data:', data);
  };

  socket.onclose = () => {
    console.log('WebSocket连接关闭');
    // 可选：添加重连逻辑，意外断开后自动恢复
    // setTimeout(() => initWebSocket(customer_id), 3000);
  };

  socket.onerror = (error) => {
    console.error('WebSocket错误:', error);
  };
}

// 初始化连接
initWebSocket(customer_id);

关键调整 #

• 把socket变量放在持久化作用域，避免重复声明
• 增加失效连接的关闭处理
• 添加错误和关闭事件监听，提升连接稳定性

二、后端支持同一用户多端连接的修改 #

原ConnectionManager用单个WebSocket实例绑定用户ID，导致多端连接被拒绝。需要改成一个用户对应多个连接的存储结构。

1. 修改ConnectionManager类 #

from typing import List, Dict
from fastapi import WebSocket

class ConnectionManager:
    def __init__(self):
        # 结构变更：key为customer_id，value为该用户的所有WebSocket连接列表
        self.connections: Dict[int, List[WebSocket]] = {}

    async def connect(self, websocket: WebSocket, customer_id: int):
        await websocket.accept()
        websocket.customer_id = customer_id
        
        # 初始化用户的连接列表（如果不存在）
        if customer_id not in self.connections:
            self.connections[customer_id] = []
        self.connections[customer_id].append(websocket)
        print(f"用户{customer_id}当前连接数: {len(self.connections[customer_id])}")

    async def disconnect(self, websocket: WebSocket):
        customer_id = getattr(websocket, 'customer_id', None)
        if customer_id in self.connections:
            # 移除当前断开的连接
            if websocket in self.connections[customer_id]:
                self.connections[customer_id].remove(websocket)
            # 连接列表为空时清理key，节省内存
            if not self.connections[customer_id]:
                del self.connections[customer_id]
            print(f"用户{customer_id}断开连接后剩余连接数: {len(self.connections.get(customer_id, []))}")

    async def broadcast(self, message: dict):
        # 给所有用户的所有连接发消息
        for customer_connections in self.connections.values():
            for websocket in customer_connections:
                await websocket.send_json(message)

    async def broadcast_to_customer(self, customer_id: int, message: dict):
        # 给该用户的所有连接发消息
        if customer_id in self.connections:
            for websocket in self.connections[customer_id]:
                await websocket.send_json(message)

connection_manager = ConnectionManager()

2. 修改WebSocket端点代码 #

移除原有的重复连接拦截逻辑，现在支持多连接：

@router.websocket("/sock")
async def websocket_endpoint(websocket: WebSocket, customer_id: int):
    try:
        await connection_manager.connect(websocket, customer_id)
        while True:
            # 保持连接存活，接收前端消息（不需要接收的话可以换成心跳逻辑）
            data = await websocket.receive_json()
            # 可选：处理前端心跳请求
            # await websocket.send_json({"type": "heartbeat", "status": "ok"})
    except WebSocketDisconnect:
        await connection_manager.disconnect(websocket)
    except Exception as e:
        print(f"WebSocket连接异常: {str(e)}")
        await connection_manager.disconnect(websocket)

三、效果验证 #

1. 前端：只会创建一个WebSocket连接，不会出现多次连接请求
2. 后端：同一用户可在电脑、移动端同时建立连接，所有连接都会收到定向推送的消息
3. 日志：后端会打印用户连接数变化，方便排查问题

内容的提问来源于stack exchange，提问作者Abrskil