在传统 Web 应用中浏览器通常通过 HTTP 向服务器发起请求服务器处理完成后返回结果。这种“请求—响应”模式非常适合查询数据、提交表单等业务但在即时聊天、实时通知、在线协作等场景中它的局限也比较明显。如果客户端需要持续获得最新数据就只能反复向服务器发送请求不仅增加服务器压力还会产生额外延迟。WebSocket 提供了一种更适合实时通信的解决方案。它可以在客户端和服务器之间建立持久连接连接建立后双方都能主动发送消息。本文将使用 FastAPI 实现一个简单的 WebSocket 服务并逐步加入连接管理、群发消息、异常处理和心跳检测等功能。一、HTTP 轮询为什么不适合实时通信假设一个网页需要实时显示服务器状态最简单的方式是每隔两秒请求一次接口setInterval(async () { const response await fetch(/api/status); const data await response.json(); console.log(data); }, 2000);这种方式被称为轮询。虽然实现简单但存在几个问题即使数据没有变化客户端也会持续发送请求实时性取决于轮询间隔轮询间隔越短服务器压力越大每次请求都需要携带 HTTP 请求头大量用户同时在线时请求数量会迅速增长。例如1000 个在线用户每两秒请求一次接口服务器每秒就要处理大约 500 个请求其中很多请求可能并没有获得新数据。WebSocket 建立连接后服务器可以只在数据发生变化时主动通知客户端从而减少无效请求。二、WebSocket 的基本工作方式WebSocket 最初通过 HTTP 发起连接请求服务器同意升级协议后连接会切换为 WebSocket。建立连接后的通信过程可以简单理解为客户端发起连接 ↓ 服务器接受连接 ↓ 双方保持长连接 ↓ 客户端和服务器随时互发消息 ↓ 任意一方关闭连接与普通 HTTP 请求相比WebSocket 主要有以下特点支持双向通信连接可以长期保持服务器能够主动推送数据单次消息的额外开销较小更适合低延迟、持续交互的业务。三、创建一个最简单的 WebSocket 服务首先安装 FastAPI 和 Uvicornpip install fastapi uvicorn创建main.pyfrom fastapi import FastAPI, WebSocket app FastAPI() app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: message await websocket.receive_text() await websocket.send_text( f服务器已收到{message} )启动服务uvicorn main:app --reload这段代码完成了三个操作接受客户端的 WebSocket 连接持续等待客户端发送文本将收到的内容返回给客户端。需要注意WebSocket 路由使用的是app.websocket()而不是普通接口使用的app.get()或app.post()。四、编写一个简单的前端客户端创建一个 HTML 页面!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleWebSocket 测试/title /head body input idmessage placeholder请输入消息 button onclicksendMessage()发送/button ul idmessages/ul script const socket new WebSocket( ws://localhost:8000/ws ); socket.onopen () { addMessage(WebSocket 已连接); }; socket.onmessage (event) { addMessage(event.data); }; socket.onclose () { addMessage(WebSocket 已断开); }; socket.onerror () { addMessage(连接发生错误); }; function sendMessage() { const input document.getElementById(message); if (!input.value.trim()) { return; } socket.send(input.value); input.value ; } function addMessage(content) { const item document.createElement(li); item.textContent content; document .getElementById(messages) .appendChild(item); } /script /body /html打开页面并建立连接后在输入框中发送消息就可以收到服务器返回的内容。生产环境通常使用 HTTPS。对应的 WebSocket 地址需要从ws://改成加密的wss://。五、管理多个客户端连接前面的代码只能处理单个客户端与服务器之间的交互。如果要实现聊天室或群发通知就需要保存当前所有在线连接。可以创建一个连接管理器from fastapi import WebSocket class ConnectionManager: def __init__(self): self.active_connections: list[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): if websocket in self.active_connections: self.active_connections.remove(websocket) async def send_personal_message( self, message: str, websocket: WebSocket, ): await websocket.send_text(message) async def broadcast(self, message: str): for connection in self.active_connections: await connection.send_text(message)然后在 WebSocket 路由中使用它from fastapi import ( FastAPI, WebSocket, WebSocketDisconnect, ) app FastAPI() manager ConnectionManager() app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: message await websocket.receive_text() await manager.broadcast( f新消息{message} ) except WebSocketDisconnect: manager.disconnect(websocket) await manager.broadcast( 一名用户已离开 )当某个用户发送消息时服务器会把消息发送给当前保存的所有连接从而实现基础的群聊效果。六、正确处理客户端断开连接网络切换、浏览器关闭、页面刷新都可能导致 WebSocket 连接中断。如果没有处理断开事件失效连接会一直保存在连接列表中。服务器下次群发消息时可能会向已经断开的连接发送数据并触发异常。FastAPI 提供了WebSocketDisconnect异常可以在客户端断开时进行清理try: while True: message await websocket.receive_text() await manager.broadcast(message) except WebSocketDisconnect: manager.disconnect(websocket)同时群发消息时也应该考虑连接可能刚好失效async def broadcast(self, message: str): disconnected [] for connection in self.active_connections: try: await connection.send_text(message) except Exception: disconnected.append(connection) for connection in disconnected: self.disconnect(connection)这样可以避免一个失效连接影响其他正常用户。七、使用 JSON 传递结构化消息实际项目中的 WebSocket 消息通常不只是简单字符串还需要携带消息类型、用户信息、时间戳和业务数据。可以使用 JSON 定义统一的消息格式{ type: chat_message, sender: user_1001, content: 你好, timestamp: 1720000000 }FastAPI 可以直接接收和发送 JSONapp.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: data await websocket.receive_json() response { type: message_received, sender: data.get(sender), content: data.get(content), } for connection in manager.active_connections: await connection.send_json(response) except WebSocketDisconnect: manager.disconnect(websocket)采用结构化消息后前端可以根据type字段执行不同操作例如chat_message显示聊天内容system_notice展示系统通知task_progress更新任务进度error提示错误信息heartbeat处理心跳检测。统一消息结构能够减少前后端沟通成本也方便以后扩展新功能。八、案例延伸同言翻译如何应用 WebSocket除了即时聊天和实时通知WebSocket 也适用于需要持续传输内容的翻译场景。以同言翻译为例在跨语言交流过程中客户端可能会连续产生需要翻译的文字或语音内容。如果每次都重新建立 HTTP 请求不仅会增加连接开销也可能影响翻译结果返回的及时性。通过 WebSocket 建立长连接后客户端可以持续发送待处理内容服务器则可以及时返回识别结果、翻译内容和任务状态。例如系统可以依次推送“内容已接收”“正在翻译”“翻译完成”等消息让用户清楚了解当前进度。对于同言翻译这类强调连续交互体验的应用WebSocket 的价值不仅是降低通信延迟还在于能够维持稳定的双向数据通道。不过完整的实时翻译系统还需要处理消息顺序、任务编号、断线重连、超时重试以及数据安全等问题。九、加入心跳检测WebSocket 虽然是长连接但连接可能因为网络设备超时、移动网络切换等原因失效。一种常见做法是让客户端定期发送心跳消息setInterval(() { if (socket.readyState WebSocket.OPEN) { socket.send(JSON.stringify({ type: ping })); } }, 30000);服务器收到后返回响应data await websocket.receive_json() if data.get(type) ping: await websocket.send_json({ type: pong }) continue服务端还可以记录每个连接最后一次心跳时间。如果某个连接长时间没有发送心跳就主动关闭并清理该连接。心跳间隔不宜设置得太短否则会产生大量无意义消息。一般可以根据业务的实时性要求和网络环境进行调整。十、生产环境需要注意的问题1. 不要只在内存中保存重要状态示例中的连接列表保存在当前进程的内存中只适合学习和单进程部署。当服务启动多个进程或部署到多台服务器时不同进程之间无法直接共享连接列表。此时可以借助 Redis Pub/Sub、消息队列等组件传递跨进程消息。2. 增加身份认证WebSocket 接口同样需要验证用户身份不能因为它不是普通 HTTP 接口就忽略安全问题。客户端可以在连接地址中携带短期令牌ws://localhost:8000/ws?tokenxxxxx服务端在接受连接前验证令牌。验证失败时应该拒绝连接。需要注意不应在日志中记录完整令牌也不建议在连接地址中传递长期有效的敏感凭证。3. 限制消息大小和发送频率如果不限制消息长度和发送频率恶意客户端可能通过发送大量数据消耗服务器资源。生产环境建议设置单条消息最大长度单个用户的发送频率最大在线连接数单个连接的空闲时间异常行为的断开策略。4. 设计断线重连机制前端检测到连接断开后可以尝试重新连接function connect() { const socket new WebSocket( ws://localhost:8000/ws ); socket.onclose () { setTimeout(connect, 3000); }; } connect();更完善的方案应该使用指数退避策略避免服务器故障时所有客户端同时高频重连。十一、WebSocket 和 HTTP 应该如何选择WebSocket 并不是 HTTP 的替代品两者适合解决不同问题。适合使用 HTTP 的场景包括普通数据查询表单提交文件上传不需要持续连接的业务对实时性要求不高的任务。适合使用 WebSocket 的场景包括即时聊天实时通知在线协作游戏状态同步任务进度推送持续产生数据的 AI 应用。一个完整系统通常会同时使用两种协议。例如用户登录、查询历史记录使用 HTTP而实时消息和状态更新使用 WebSocket。十二、总结使用 FastAPI 构建一个基础 WebSocket 服务并不复杂但从演示代码走向生产环境还需要处理很多工程问题多客户端连接管理断开连接清理JSON 消息协议心跳检测身份认证发送频率限制断线重连多进程消息同步。WebSocket 的核心价值是让服务器和客户端在一条持久连接上进行双向通信。对于即时聊天、实时通知和连续 AI 处理等业务它可以提供比传统轮询更及时、更高效的交互体验。在技术选型时不必为了“实时”而盲目使用 WebSocket。如果业务只需要偶尔获取一次数据普通 HTTP 通常更加简单可靠只有在客户端和服务器需要持续交换信息时WebSocket 的优势才会真正体现出来。