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()<br>online_count = 0<br><br>@app.websocket("/ws")<br>async def websocket_endpoint(websocket: WebSocket):<br> await websocket.accept()<br> conn_id = id(websocket)<br> connections.add(conn_id)<br> online_count = len(connections) # 注意：这不是原子操作<br> try:<br> while True:<br> await websocket.receive_text()<br> finally:<br> 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是唯一合理选择。
注意两点：一是连接池复用，别每次操作都新建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 = []<br><br>async def broadcast_count(count: int):<br> for ws in active_sockets[:]: # 遍历副本防修改中迭代<br> try:<br> await ws.send_json({"type": "count", "value": count})<br> except (WebSocketDisconnect, RuntimeError):<br> active_sockets.remove(ws) - Redis Pub/Sub方案：一个服务监听
online:update频道发消息，所有FastAPI实例订阅并转发给本地连接——适合横向扩展场景
真实项目里，连接数超过几百后，广播性能会明显下降；这时该考虑用消息队列（如NATS）或专用实时服务（如Ably），而不是硬撑。
大量免费API接口：立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游，为数据API应用能力赋能！
