Python开发FastAPI怎么实现WebSocket在线人数统计_基于内存字典或Redis维护连接池

FastAPI中需用可哈希标识（如id(websocket)或user_id）增减在线人数，内存方案须加锁防竞态，多进程必须用Redis原子指令incr/decr维护共享计数，并配合心跳键实现高可用。

WebSocket连接建立时怎么安全地增减在线人数

FastAPI的WebSocket实例没有唯一ID，直接用websocket对象做字典键会报TypeError: unhashable type: 'WebSocket'。必须提取可哈希标识——推荐用id(websocket)（内存方案）或客户端传来的user_id（业务友好）。Redis方案则建议用带前缀的字符串键，比如ws:conn:{user_id}，避免键冲突。

增减逻辑必须包裹在try/finally里：连接建立时add，断开时remove，否则异常中断会导致计数漂移。别依赖websocket.close()回调——它不总触发，尤其客户端强制断网时。

内存方案示例：

connections = set()
online_count = 0

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    conn_id = id(websocket)
    connections.add(conn_id)
    online_count = len(connections)  # 注意：这不是原子操作
    try:
        while True:
            await websocket.receive_text()
    finally:
        connections.discard(conn_id)

更稳妥的内存计数应改用threading.Lock或asyncio.Lock保护online_count变量

为什么不能直接用全局变量+len()统计在线人数

多个协程并发修改同一整数变量（如online_count += 1）会产生竞态条件——+=不是原子操作，实际是读-改-写三步，中间可能被其他协程打断。结果就是上线10人，print(online_count)却只显示6或8。

Redis方案看似绕过这个问题，但也要注意：INCR/DECR本身是原子的，可直接用；但先GET再SET就不是。所以必须用redis.incr("online_count")和redis.decr("online_count")，而不是redis.get("online_count")后自己加减再set。

错误写法：count = int(redis.get("online_count") or 0); redis.set("online_count", count + 1)
正确写法：redis.incr("online_count")（上线）和redis.decr("online_count")（下线）
如果需存储连接详情（如用户IP、登录时间），用HSET ws:conn:{id} ip {ip} ts {time.time()}，配合KEYS ws:conn:*查总数（不推荐生产用）或单独维护计数键

如何让不同FastAPI进程共享WebSocket在线数

单进程用内存字典可行，但gunicorn多worker或K8s多副本部署时，各进程内存隔离，online_count变成各自为政。此时必须外置状态存储——Redis是唯一合理选择。

Python 3.14.3

微软官方的 Python 扩展，是 VS Code 安装量最高的扩展（209M+）。集成 IntelliSense（通过 Pylance）、调试（通过 Python Debugger）、代码检查、格式化、重构和单元测试等功能。支持 Jupyter Notebook、虚拟环境管理和多 Python 版本切换。

注意两点：一是连接池复用，别每次操作都新建redis.Redis()实例；二是设置合理的键过期时间（EXPIRE），防止客户端异常断连后残留。例如在websocket.accept()后立刻执行redis.setex("ws:heartbeat:{conn_id}", 60, "alive")，再配后台任务定期清理超时键。

推荐初始化一次连接：

redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

上线：redis_client.incr("online_total") + redis_client.setex(f"ws:active:{id(websocket)}", 300, "1")
下线：redis_client.decr("online_total") + redis_client.delete(f"ws:active:{id(websocket)}")

怎么实时把人数变化推给所有客户端

FastAPI没内置广播机制，得自己维护连接列表。内存方案可用list[WebSocket]存活跃连接，但要注意：WebSocket对象不可序列化，不能存进Redis；且发送时需捕获WebSocketDisconnect并主动清理。

广播逻辑别塞进主WebSocket循环——它会阻塞当前连接的消息处理。应该另起一个后台任务（asyncio.create_task()）或用发布订阅模式（Redis Pub/Sub）解耦。

简单广播示例：

active_sockets = []

async def broadcast_count(count: int):
    for ws in active_sockets[:]:  # 遍历副本防修改中迭代
        try:
            await ws.send_json({"type": "count", "value": count})
        except (WebSocketDisconnect, RuntimeError):
            active_sockets.remove(ws)