1. 什么是 WebSocket
WebSocket 是一种在单个 TCP 连接上进行全双工（Full-Duplex）通信的协议。它允许服务器主动向客户端推送数据，打破了传统 Web 通信中“客户端不问、服务器不答”的限制。
在 WebSocket 出现之前，网页如果想实时获取服务器的新数据（比如聊天消息、股票行情、实时比分），主要依靠以下两种方法：
轮询（Polling），即客户端每隔一段时间（如几秒钟）向服务器发一次 HTTP 请求，其具体过程如下所示：
这种方案最大的问题是浪费带宽和服务器资源，因为绝大多数的请求都没办法获得最终的数据，轮询需要不断的发起连接，断开连接；除此之外，我们需要每隔一段时间发起一次请求，时间间隔的确定对于部分对实时性要求比较高的业务也是非常不友好的，因此没有办法很好判断服务端在什么时候已经 ready。轮训方案也并非一无是处，最大的好处就是开发简单，直接发起 HTTP 请求即可。
为了解决上面的两个问题，这便有了长轮询（Long Polling），即客户端发一个请求到服务器，服务器把连接挂起，直到有新数据才返回，返回后客户端立刻再发起下一个请求，其具体过程如下图所示：
相比较于轮询方案来说，长轮询减少了客户端发起 H TTP 连接的开销，改成在服务器端主动地去判断关注的内容是否变化。虽然减少了无效的请求，但如果多次请求，依然存在连接频繁断开重连的开销，且在服务端维持连接会消耗服务器内存。
WebSocket 方案正是为解决这些问题而来，在 WebSocket 方案中，只需要一次握手，双方便建立一条持久性的连接，可以随时互发信息，其具体过程如下图所示：
WebSocket 的信息交互只需要建立一次连接。
2. WebSocket 的基本原理
2.1. 工作原理
WebSocket 的生命周期分为两个阶段：握手阶段（Handshake）和数据传输阶段（Data Transfer）。
2.1.1. 握手阶段（Handshake）
在握手阶段，WebSocket 借助 HTTP 协议完成客户端和服务端的握手，具体而言，
- 客户端向服务器发送一个特殊的 HTTP 请求，头部包含
Upgrade: websocket和Sec-WebSocket-Key等字段，表示：“我想把协议升级为 WebSocket”。 - 服务器收到后，如果同意，会返回一个状态码为
101 Switching Protocols的响应。 - 握手成功！之后，底层的 TCP 连接被保持，双方正式切换到 WebSocket 协议通信，不再使用 HTTP。
2.1.2. 数据传输阶段（Data Transfer）
- 全双工： 客户端可以随时发信给服务器，服务器也可以随时发信给客户端，互不干扰，真正实现了实时数据双向传输。
- 轻量级数据帧： 相比于 HTTP 每次都要携带庞大的 Header（Cookie、User-Agent 等），WebSocket 的数据包头部非常小（通常只有 2-10 字节），大大减少了网络开销。
2.2. 基于 FastAPI 的 WebSocket 实现
FastAPI 是一个卓越的基于 Python 的服务端框架，提供了 WebSocket 的实现，我们利用其写一段服务端的代码，代码如下：
import uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
@app.websocket("/test_websocket")
async def websocket_endpoint(websocket: WebSocket):
# 接受连接
await websocket.accept()
try:
while True:
# 接收消息
data = await websocket.receive_text()
# 发送消息
await websocket.send_text(data)
except WebSocketDisconnect:
print("断开连接")
if __name__ == "__main__":
uvicorn.run('main:app', host='0.0.0.0', port=8001)
上述代码，websocket.accept() 接受连接，websocket.receive_text() 接收客户端的消息，websocket.send_text(data) 向客户端发送信息。
再写个简单的客户端，代码如下：
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>WebSocket 测试</title>
</head>
<body>
<div id="connect">
<button onclick="wsConnect()">连接 WebSocket 服务器</button>
</div>
<div class="send_area">
<span>发送信息：<input type="text" id="input_a" /></span>
<button onclick="sendMsg()">发送</button>
</div>
<div class="receive_area">
<textarea id="receive_a"></textarea>
</div>
<script>
let wsClient;
function wsConnect() {
wsClient = new WebSocket("ws://localhost:8001/test_websocket");
wsClient.onopen = function() {
document.getElementById("receive_a").value = "连接成功";
}
wsClient.onmessage = function(event) {
console.log(event)
let msg = "\n" + event.data
document.getElementById("receive_a").value += msg
}
wsClient.onclose = function() {
document.getElementById("receive_a").value = "\n断开连接";
}
}
function sendMsg() {
let msg = document.getElementById("input_a");
console.log(msg.value)
wsClient.send(msg.value);
}
</script>
</body>
</html>
一个简简单单的基于 WebSocket 的客户端和服务端通信就构建完成了，这时候启动服务端，同时打开客户端页面，就可以发送信息了，如下图：
2.3. 实现简单的聊天室
上述也就是简单的验证了原理，现在我们稍微复杂点，实现个聊天室的功能，假设 A 用户发送信息，信息需要触达到用户 B，C 等。首先对 WebSocket 的服务端改造，让其能够记录多个用户，代码如下：
import uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
app = FastAPI()
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):
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, exclude_self: bool = True, sender: WebSocket = None):
"""广播消息给所有连接，可选择是否排除发送者"""
for connection in self.active_connections:
if exclude_self and connection == sender:
continue
try:
await connection.send_text(message)
except:
# 如果发送失败（连接已断），从列表中移除
self.disconnect(connection)
manager = ConnectionManager()
@app.websocket("/test_websocket")
async def websocket_endpoint(websocket: WebSocket):
# 接受连接
await manager.connect(websocket)
try:
while True:
# 接收消息
data = await websocket.receive_text()
print(data)
# 发送消息
await manager.broadcast(data, exclude_self=True, sender=websocket)
except WebSocketDisconnect:
print("断开连接")
if __name__ == "__main__":
uvicorn.run('main:app', host='0.0.0.0', port=8001)
直接起多个客户端，就可以相互发送信息了。
3. 应用场景
综上，WebSocket 非常适合需要高实时性、频繁双向交互的业务场景：
- 即时通讯（Chat）： 如微信网页版、钉钉、各类聊天室，消息收发必须实时且流畅。
- 实时看板与数据监控： 股票行情走势图、服务器运维监控大屏、双十一实时交易大屏。
- 协同编辑： 如石墨文档、腾讯文档或 Figma，多个人同时编辑一个文档，每个人的修改要瞬间同步给其他人。
- 在线游戏： 网页端或小游戏中的实时对战数据同步。
- 消息通知： 网站的实时消息提示、弹幕系统。